基于Python打造一個(gè)可視化FTP服務(wù)器

更新時(shí)間：2025年04月08日 14:11:11 作者：探客白澤

在日常辦公和團(tuán)隊(duì)協(xié)作中,文件共享是一個(gè)不可或缺的需求,所以本文將使用Python + Tkinter + pyftpdlib 開發(fā)一款可視化FTP服務(wù)器,有需要的小伙伴可以參考下

1. 概述

在日常辦公和團(tuán)隊(duì)協(xié)作中，文件共享是一個(gè)不可或缺的需求。然而，市場(chǎng)上的FTP服務(wù)器軟件通常配置復(fù)雜、體積龐大，或者收費(fèi)昂貴。有沒有一種簡(jiǎn)單高效的方式，讓你可以輕松搭建自己的FTP服務(wù)器，實(shí)現(xiàn)高效的文件共享呢？

答案就是——使用 Python + Tkinter + pyftpdlib 開發(fā)一款可視化FTP服務(wù)器，讓你無需命令行操作，也能輕松管理FTP服務(wù)器！本文將詳細(xì)介紹這款FTP服務(wù)器的功能、實(shí)現(xiàn)原理，以及如何使用它來搭建屬于自己的文件共享中心。

2. 功能介紹

該FTP服務(wù)器具備以下核心功能：

GUI 操作：基于 Tkinter 圖形化界面，無需命令行操作，所有設(shè)置一目了然。

多IP支持：自動(dòng)檢測(cè)本機(jī)有效IP地址，支持局域網(wǎng)和公網(wǎng)訪問。

多用戶管理：支持添加多個(gè)用戶，分配不同的訪問權(quán)限。

權(quán)限控制：提供詳細(xì)的權(quán)限配置（上傳、下載、刪除、重命名等）。

實(shí)時(shí)連接監(jiān)控：顯示當(dāng)前在線用戶數(shù)量，連接、斷開的日志實(shí)時(shí)更新。

目錄選擇：支持自定義FTP根目錄，靈活管理共享文件。

操作日志：日志窗口實(shí)時(shí)記錄服務(wù)器狀態(tài)，便于調(diào)試和管理。

3. 如何使用

1.運(yùn)行程序

首先，你需要確保已經(jīng)安裝了 Python 環(huán)境，并安裝了 pyftpdlib 依賴包。

pip install pyftpdlib

然后運(yùn)行 FTPServerGUI.py 文件，啟動(dòng)FTP服務(wù)器圖形化界面。

python FTPServerGUI.py

2.選擇FTP目錄

在軟件界面中，點(diǎn)擊 “選擇FTP目錄” 按鈕，選擇你要共享的文件夾。

該目錄即為FTP服務(wù)器的根目錄，所有FTP用戶只能在該目錄及其子目錄下操作。

3.配置用戶信息

輸入用戶名和密碼，確保用戶可以安全訪問服務(wù)器。

4.設(shè)定訪問權(quán)限

勾選用戶權(quán)限：

上傳 (w)：允許用戶上傳文件
下載 ®：允許用戶下載文件
刪除 (d)：允許用戶刪除文件
創(chuàng)建目錄 (m)：允許用戶在FTP目錄下創(chuàng)建文件夾
修改權(quán)限 (M)：允許用戶更改文件或文件夾權(quán)限

根據(jù)實(shí)際需求，自由組合用戶的訪問權(quán)限。

5.啟動(dòng)服務(wù)器

點(diǎn)擊 “啟動(dòng)服務(wù)器” 按鈕，F(xiàn)TP服務(wù)器即刻運(yùn)行。

此時(shí)，日志窗口會(huì)顯示服務(wù)器狀態(tài)，并提供可用的FTP訪問地址。

[狀態(tài)] 服務(wù)已啟動(dòng)于 0.0.0.0:21
[提示] 可用地址: 192.168.1.100

6.連接FTP服務(wù)器

你可以使用 Windows 資源管理器、FileZilla、或者命令行連接FTP服務(wù)器。

方式1：Windows 資源管理器

在地址欄輸入：

ftp://192.168.1.100

然后輸入用戶名和密碼，即可訪問FTP文件。

方式2：FileZilla

使用 FileZilla 連接服務(wù)器，輸入服務(wù)器IP、端口（21）、用戶名、密碼，即可管理文件。

方式3：命令行FTP

ftp 192.168.1.100

輸入用戶名和密碼后，可以使用 FTP 命令操作文件。

ls   # 列出文件
get 文件名  # 下載文件
put 文件名  # 上傳文件
exit  # 退出FTP

4. 代碼解析

FTP服務(wù)器的核心功能基于 pyftpdlib 實(shí)現(xiàn)，以下是關(guān)鍵代碼解析：

1.服務(wù)器啟動(dòng)邏輯

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): 監(jiān)聽 0.0.0.0，接受所有IP訪問。

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

2.用戶權(quán)限管理

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

perm_vars 存儲(chǔ)用戶權(quán)限，如 w（寫）、r（讀）、d（刪除）等。

通過 add_user 方法添加用戶及其權(quán)限。

3.連接狀態(tài)監(jiān)控

def update_connection_count(self, delta: int):
    self.connection_count += delta
    self.root.after(0, self.conn_label.config, {"text": f"當(dāng)前連接: {self.connection_count}"})

update_connection_count 線程安全地更新當(dāng)前在線用戶數(shù)量。

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

5. 運(yùn)行效果

6.相關(guān)源碼

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):
        # 主窗口設(shè)置
        self.root.title("FTP文件共享服務(wù)器")
        self.root.geometry("560x500")
        self.root.minsize(560, 480)

        # 配置網(wǎng)格布局權(quán)重
        self.root.columnconfigure(0, weight=1)
        self.root.columnconfigure(1, weight=1)
        self.root.rowconfigure(4, weight=1)  # 日志區(qū)域行

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

        # 當(dāng)前連接數(shù)顯示
        self.conn_label = ttk.Label(top_info, text="當(dāng)前連接: 0", foreground="green")
        self.conn_label.pack(side="right", padx=10)

        # IP地址顯示
        ttk.Label(top_info, text="服務(wù)器IP:").pack(side="left")
        self.ip_label = ttk.Label(top_info, text="正在獲取...", foreground="blue")
        self.ip_label.pack(side="left", padx=5)

        # 配置框架
        config_frame = ttk.LabelFrame(self.root, text="服務(wù)器配置")
        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)

        # 權(quán)限設(shè)置框架
        perm_frame = ttk.LabelFrame(self.root, text="用戶權(quán)限設(shè)置")
        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)

        # 權(quán)限項(xiàng)分布
        permissions = [
        ("上傳文件 (w)", 'w', 0),
        ("下載文件 (r)", 'r', 0),
        ("刪除文件 (d)", 'd', 0),
        ("查看列表 (l)", 'l', 1),
        ("切換目錄 (e)", 'e', 1),
        ("創(chuàng)建目錄 (m)", 'm', 1),
        ("修改權(quán)限 (M)", 'M', 2),
        ("重命名 (f)", 'f', 2),
            ("追加文件 (a)", 'a', 2)
        ]

        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="啟動(dòng)服務(wù)器", command=self.toggle_server)
        self.start_btn.pack(side="left", padx=5)

        # 日志區(qū)域
        log_frame = ttk.LabelFrame(self.root, text="服務(wù)器日志")
        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"))

        # 布局權(quán)重配置
        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默認(rèn)地址
            return False
        if ip.startswith("192.168.") or ip.startswith("10."):  # 內(nèi)網(wǎng)地址
            return True  # 保留顯示內(nèi)網(wǎng)地址
        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):
        """線程安全更新連接數(shù)"""
        self.connection_count += delta
        self.root.after(0, self.conn_label.config, 
                      {"text": f"當(dāng)前連接: {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("錯(cuò)誤", "請(qǐng)?zhí)顚懰斜靥钭侄危?)
            return

        perm = ''.join([k for k, v in self.perm_vars.items() if v.get()])
        if not perm:
            messagebox.showerror("錯(cuò)誤", "請(qǐng)至少選擇一個(gè)權(quán)限！")
            return

        class CustomHandler(FTPHandler):
            gui = self  # 引用GUI實(shí)例

            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"[認(rèn)證] 用戶 {username} 登錄成功")

            def on_login_failed(self, username, password):
                self.gui.thread_safe_log(f"[認(rèn)證] 登錄失敗 用戶名: {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"[狀態(tài)] 服務(wù)已啟動(dòng)于 0.0.0.0:{port}")
            self.thread_safe_log(f"[提示] 可用地址: {self.ip_label.cget('text')}")
            self.start_btn.config(text="停止服務(wù)器")

            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"[錯(cuò)誤] 啟動(dòng)失敗: {str(e)}")

    def stop_server(self):
        if self.server:
            self.server.close_all()
            self.server = None
            self.connection_count = 0
            self.conn_label.config(text="當(dāng)前連接: 0")
            self.thread_safe_log("[狀態(tài)] 服務(wù)器已停止")
            self.start_btn.config(text="啟動(dòng)服務(wù)器")

    def display_welcome_msg(self):
        """顯示歡迎信息"""
        welcome_text = """=== FTP文件共享服務(wù)器 ===
【使用指南】
1. 選擇FTP目錄 -> 設(shè)置用戶權(quán)限 -> 啟動(dòng)服務(wù)
2. 客戶端連接地址顯示在頂部IP欄
3. 日志區(qū)域?qū)崟r(shí)顯示連接狀態(tài)
【常用命令幫助】
USER [用戶名]   - 登錄認(rèn)證
PASS [密碼]     - 輸入密碼
LIST            - 列出文件列表
RETR [文件名]   - 下載文件
STOR [文件名]   - 上傳文件
QUIT/EXIT       - 斷開連接
當(dāng)前版本功能：
? 多IP地址自動(dòng)檢測(cè)
? 實(shí)時(shí)連接數(shù)統(tǒng)計(jì)
? 細(xì)粒度權(quán)限控制
? 操作日志記錄
"""
        self.log_text.configure(state='normal')
        self.log_text.delete(1.0, tk.END)
        self.log_text.insert(tk.END, welcome_text)

        # 設(shè)置文本樣式
        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()