PyQt 实现FTP客户端调试助手源码分享

PyQt 实现FTP客户端调试助手源码分享

• 一、效果展示
• 二、源码分享
• • 1、工程结构
  • 2、库安装
  • 3、main.py
  • 4、ftp_server_test.py（测试使用）
• 三、实现原理
• • 1、协议拦截：继承 FTP 重写 putline / getline
  • 2、线程模型：QThread + 任务队列
  • • 2.1 任务队列
    • 2.2 线程安全
  • 3、信号槽通信：FtpSignals
  • 4、目录列表解析：MLSD 优先，LIST 兜底
  • 5、上传/下载进度跟踪
  • • 5.1 上传：包装文件对象
    • 5.2 下载：回调函数
  • 6、原始命令执行
  • 7、整体数据流
  • 8、小结

一、效果展示

二、源码分享

1、工程结构

2、库安装

3、main.py

"""
功能：
– 连接/断开 FTP 服务器
– 本地/远程双栏文件浏览器
– 文件上传/下载（带进度）
– 调试控制台：实时 FTP 命令/响应日志 + 手动发送原始命令
"""

import os
import sys
import threading
import time
from datetime import datetime
from ftplib import FTP, error_perm

from PyQt6.QtCore import (
Qt, QThread, pyqtSignal, QObject, QSettings
)
from PyQt6.QtGui import (
QFont, QColor, QPalette
)
from PyQt6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QGridLayout, QGroupBox, QLabel, QLineEdit, QPushButton,
QTreeWidget, QTreeWidgetItem, QHeaderView, QTextEdit,
QSplitter, QProgressBar, QFileDialog, QMessageBox,
QStatusBar, QToolBar, QMenu, QAbstractItemView, QComboBox
)

# ============================================================
# 调试 FTP 客户端 – 拦截所有命令和响应
# ============================================================
class DebugFTP(FTP):
"""继承 ftplib.FTP，重写 putline/getline 以捕获所有协议交互"""

def __init__(self, *args, log_callback=None, **kwargs):
self._log_callback = log_callback
super().__init__(*args, **kwargs)

def putline(self, line):
if self._log_callback:
self._log_callback("cmd", line)
super().putline(line)

def getline(self):
line = super().getline()
if self._log_callback:
self._log_callback("resp", line)
return line

# ============================================================
# FTP 信号定义
# ============================================================
class FtpSignals(QObject):
log_message = pyqtSignal(str, str) # (类别, 消息) 类别: cmd/resp/info/error/success
file_list_received = pyqtSignal(list, str) # (文件列表, 当前路径)
transfer_progress = pyqtSignal(str, int, int) # (文件名, 已传输, 总大小)
transfer_complete = pyqtSignal(str, bool, str) # (文件名, 成功, 消息)
connection_changed = pyqtSignal(bool) # 连接状态
raw_command_result = pyqtSignal(str) # 原始命令执行结果

# ============================================================
# FTP 工作线程
# ============================================================
class FtpWorkerThread(QThread):
"""在后台线程执行所有 FTP 操作"""

def __init__(self, parent=None):
super().__init__(parent)
self.signals = FtpSignals()
self.ftp = None
self._connected = False
self._stop_flag = False
self._task_queue = []
self._lock = threading.Lock()

def connect_to_server(self, host, port, user, password):
"""添加连接任务"""
self._enqueue("connect", {
"host": host, "port": port,
"user": user, "password": password
})

def disconnect_from_server(self):
"""添加断开连接任务"""
self._enqueue("disconnect", {})

def list_directory(self, path=""):
"""添加列目录任务"""
self._enqueue("list", {"path": path})

def change_directory(self, path):
"""添加切换目录任务"""
self._enqueue("cwd", {"path": path})

def upload_file(self, local_path, remote_name):
"""添加上传任务"""
self._enqueue("upload", {
"local_path": local_path, "remote_name": remote_name
})

def download_file(self, remote_name, local_path):
"""添加下载任务"""
self._enqueue("download", {
"remote_name": remote_name, "local_path": local_path
})

def execute_raw_command(self, command):
"""添加原始命令执行任务"""
self._enqueue("raw_cmd", {"command": command})

def delete_file(self, name):
"""添加删除文件任务"""
self._enqueue("delete_file", {"name": name})

def delete_directory(self, name):
"""添加删除目录任务"""
self._enqueue("delete_dir", {"name": name})

def _enqueue(self, task_type, params):
with self._lock:
self._task_queue.append((task_type, params))

def run(self):
while not self._stop_flag:
task = None
with self._lock:
if self._task_queue:
task = self._task_queue.pop(0)

if task is None:
self.msleep(50)
continue

task_type, params = task
try:
if task_type == "connect":
self._do_connect(**params)
elif task_type == "disconnect":
self._do_disconnect()
elif task_type == "list":
self._do_list(**params)
elif task_type == "cwd":
self._do_cwd(**params)
elif task_type == "upload":
self._do_upload(**params)
elif task_type == "download":
self._do_download(**params)
elif task_type == "raw_cmd":
self._do_raw_command(**params)
elif task_type == "delete_file":
self._do_delete_file(**params)
elif task_type == "delete_dir":
self._do_delete_dir(**params)
except Exception as e:
self.signals.log_message.emit("error", f"操作失败: {str(e)}")

def stop(self):
self._stop_flag = True
if self.ftp and self._connected:
try:
self.ftp.quit()
except Exception:
pass
self.wait(3000)

def _do_connect(self, host, port, user, password):
self.signals.log_message.emit("info", f"正在连接 {host}:{port} …")
try:
self.ftp = DebugFTP(log_callback=self._ftp_log)
self.ftp.connect(host, int(port), timeout=10)
self.ftp.login(user, password)
self._connected = True
self.signals.log_message.emit("success", f"已连接到 {host}:{port}")
self.signals.connection_changed.emit(True)
# 连接成功后列出根目录
self._do_list(path="")
except Exception as e:
self.signals.log_message.emit("error", f"连接失败: {str(e)}")
self.signals.connection_changed.emit(False)
self._connected = False

def _do_disconnect(self):
try:
if self.ftp:
self.ftp.quit()
except Exception:
pass
self._connected = False
self.ftp = None
self.signals.log_message.emit("info", "已断开连接")
self.signals.connection_changed.emit(False)

def _do_list(self, path=""):
if not self._connected or not self.ftp:
return

try:
if path:
self.ftp.cwd(path)

cwd = self.ftp.pwd()
self.signals.log_message.emit("info", f"当前目录: {cwd}")

items = []
try:
# 尝试 MLSD（结构化目录列表）
for name, facts in self.ftp.mlsd():
if name == ".":
continue
if name == "..":
items.append(("..", 0, True, ""))
continue
is_dir = facts.get("type", "") == "dir"
size = int(facts.get("size", 0))
modify = facts.get("modify", "")
items.append((name, size, is_dir, modify))
except error_perm:
# 回退到 LIST 命令解析
lines = []
self.ftp.retrlines("LIST", lines.append)
for line in lines:
parsed = self._parse_list_line(line)
if parsed:
items.append(parsed)

self.signals.file_list_received.emit(items, cwd)

except Exception as e:
self.signals.log_message.emit("error", f"列目录失败: {str(e)}")

def _do_cwd(self, path):
if not self._connected or not self.ftp:
return
try:
self.ftp.cwd(path)
self._do_list()
except Exception as e:
self.signals.log_message.emit("error", f"切换目录失败: {str(e)}")

def _do_upload(self, local_path, remote_name):
if not self._connected or not self.ftp:
return
filename = os.path.basename(local_path)
file_size = os.path.getsize(local_path)
self.signals.log_message.emit("info", f"开始上传: {filename} ({self._format_size(file_size)})")

transferred = [0]
last_emit = [time.time()]

class TrackingFile:
"""包装文件对象，在每次 read 时跟踪传输进度"""
def __init__(self, fobj):
self._f = fobj
def read(self, *args, **kwargs):
data = self._f.read(*args, **kwargs)
if data:
transferred[0] += len(data)
now = time.time()
if now – last_emit[0] > 0.1:
self.signals.transfer_progress.emit(
filename, transferred[0], file_size
)
last_emit[0] = now
return data
def __getattr__(self, name):
return getattr(self._f, name)

try:
with open(local_path, "rb") as f:
self.ftp.storbinary(f"STOR {remote_name}", TrackingFile(f))

self.signals.transfer_progress.emit(filename, file_size, file_size)
self.signals.transfer_complete.emit(filename, True, "上传成功")
self.signals.log_message.emit("success", f"上传完成: {filename}")
except Exception as e:
self.signals.transfer_complete.emit(filename, False, str(e))
self.signals.log_message.emit("error", f"上传失败: {filename} – {str(e)}")

def _do_download(self, remote_name, local_path):
if not self._connected or not self.ftp:
return
filename = os.path.basename(remote_name)
self.signals.log_message.emit("info", f"开始下载: {filename}")

# 获取文件大小
file_size = 0
try:
file_size = self.ftp.size(remote_name) or 0
except Exception:
pass

transferred = [0]
last_emit = [time.time()]

def on_data(data):
f.write(data)
transferred[0] += len(data)
now = time.time()
if now – last_emit[0] > 0.1:
self.signals.transfer_progress.emit(filename, transferred[0], file_size)
last_emit[0] = now

try:
with open(local_path, "wb") as f:
self.ftp.retrbinary(f"RETR {remote_name}", on_data)

if file_size > 0:
self.signals.transfer_progress.emit(filename, file_size, file_size)
self.signals.transfer_complete.emit(filename, True, "下载成功")
self.signals.log_message.emit("success", f"下载完成: {filename}")
except Exception as e:
self.signals.transfer_complete.emit(filename, False, str(e))
self.signals.log_message.emit("error", f"下载失败: {filename} – {str(e)}")

def _do_raw_command(self, command):
if not self._connected or not self.ftp:
self.signals.raw_command_result.emit("错误: 未连接到服务器")
return

try:
cmd = command.strip()
if not cmd:
return

parts = cmd.split(None, 1)
cmd_name = parts[0].upper()
cmd_args = parts[1] if len(parts) > 1 else ""

# 特殊处理一些常见命令
if cmd_name == "LIST":
lines = []
if cmd_args:
self.ftp.retrlines(f"LIST {cmd_args}", lines.append)
else:
self.ftp.retrlines("LIST", lines.append)
result = "\\n".join(lines) if lines else "(空目录)"
elif cmd_name == "NLST":
lines = []
self.ftp.retrlines(f"NLST {cmd_args}" if cmd_args else "NLST", lines.append)
result = "\\n".join(lines) if lines else "(空)"
elif cmd_name == "PWD":
result = self.ftp.pwd()
elif cmd_name == "CWD":
self.ftp.cwd(cmd_args)
result = f"当前目录: {self.ftp.pwd()}"
elif cmd_name == "SIZE":
size = self.ftp.size(cmd_args)
result = f"{cmd_args}: {size} bytes" if size is not None else "无法获取大小"
elif cmd_name == "MKD":
result = self.ftp.mkd(cmd_args)
elif cmd_name == "RMD":
self.ftp.rmd(cmd_args)
result = f"已删除目录: {cmd_args}"
elif cmd_name == "DELE":
self.ftp.delete(cmd_args)
result = f"已删除文件: {cmd_args}"
elif cmd_name == "REN":
# REN old new
ren_parts = cmd_args.split(None, 1)
if len(ren_parts) == 2:
self.ftp.rename(ren_parts[0], ren_parts[1])
result = f"已重命名: {ren_parts[0]} -> {ren_parts[1]}"
else:
result = "用法: REN <旧名> <新名>"
elif cmd_name == "HELP":
resp = self.ftp.sendcmd("HELP")
result = resp
elif cmd_name in ("QUIT", "BYE"):
result = "请使用界面上的断开按钮"
else:
# 通用命令发送
resp = self.ftp.sendcmd(cmd)
result = resp

self.signals.raw_command_result.emit(result)

except Exception as e:
self.signals.raw_command_result.emit(f"错误: {str(e)}")

def _do_delete_file(self, name):
"""删除远程文件"""
if not self._connected or not self.ftp:
return
try:
self.ftp.delete(name)
self.signals.log_message.emit("success", f"已删除文件: {name}")
# 删除后刷新目录
self._do_list()
except Exception as e:
self.signals.log_message.emit("error", f"删除文件失败: {name} – {str(e)}")

def _do_delete_dir(self, name):
"""删除远程目录"""
if not self._connected or not self.ftp:
return
try:
self.ftp.rmd(name)
self.signals.log_message.emit("success", f"已删除目录: {name}")
# 删除后刷新目录
self._do_list()
except Exception as e:
self.signals.log_message.emit("error", f"删除目录失败: {name} – {str(e)}")

def _ftp_log(self, category, message):
"""FTP 命令/响应日志回调"""
self.signals.log_message.emit(category, message)

def _parse_list_line(self, line):
"""解析 LIST 命令输出的每一行"""
# Unix 格式: drwxr-xr-x 2 user group 4096 Jan 01 12:00 filename
# DOS 格式: 01-01-24 12:00AM <DIR> dirname
# 01-01-24 12:00AM 1234 filename
parts = line.split(None, 8)

if len(parts) >= 9 and (parts[0][0] in ('d', '-', 'l', 'r')):
# Unix 格式
is_dir = parts[0][0] == 'd'
try:
size = int(parts[4])
except (ValueError, IndexError):
size = 0
name = parts[8]
return (name, size, is_dir, "")

elif len(parts) >= 4:
# 尝试 DOS 格式
try:
is_dir = parts[2] == '<DIR>'
size = 0 if is_dir else int(parts[2])
name = parts[3]
return (name, size, is_dir, "")
except (ValueError, IndexError):
pass

return None

@staticmethod
def _format_size(size_bytes):
if size_bytes < 1024:
return f"{size_bytes} B"
elif size_bytes < 1024 * 1024:
return f"{size_bytes / 1024:.1f} KB"
elif size_bytes < 1024 * 1024 * 1024:
return f"{size_bytes / (1024 * 1024):.1f} MB"
else:
return f"{size_bytes / (1024 * 1024 * 1024):.2f} GB"

# ============================================================
# 文件浏览器 Widget
# ============================================================
class FileBrowserWidget(QWidget):
"""文件浏览器组件，用于远程和本地文件浏览"""

file_selected = pyqtSignal(str, bool) # (文件路径, 是否为目录)
file_delete_requested = pyqtSignal(str, bool) # (名称, 是否为目录)

def __init__(self, title="文件浏览器", is_remote=False, parent=None):
super().__init__(parent)
self.is_remote = is_remote
self.current_path = ""
self.path_history = []

layout = QVBoxLayout(self)
layout.setContentsMargins(4, 4, 4, 4)
layout.setSpacing(4)

# 标题和路径
header = QHBoxLayout()
self.title_label = QLabel(f"<b>{title}</b>")
header.addWidget(self.title_label)
header.addStretch()

self.path_label = QLineEdit()
self.path_label.setReadOnly(True)
self.path_label.setPlaceholderText("路径…")
self.path_label.setMinimumWidth(200)
header.addWidget(self.path_label, stretch=1)

layout.addLayout(header)

# 文件列表
self.tree = QTreeWidget()
self.tree.setHeaderLabels(["名称", "大小", "类型"])
self.tree.setColumnCount(3)
self.tree.setRootIsDecorated(False)
self.tree.setSortingEnabled(True)
self.tree.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self.tree.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self.tree.customContextMenuRequested.connect(self._show_context_menu)
self.tree.itemDoubleClicked.connect(self._on_item_double_clicked)

header_view = self.tree.header()
header_view.setStretchLastSection(True)
header_view.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
header_view.setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents)
header_view.setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents)

layout.addWidget(self.tree)

def set_file_list(self, items, current_path=""):
"""设置文件列表 items: [(name, size, is_dir, modify), …]"""
self.tree.clear()
self.current_path = current_path
self.path_label.setText(current_path)

# 排序：目录在前，文件在后
dirs = sorted([i for i in items if i[2]], key=lambda x: x[0].lower())
files = sorted([i for i in items if not i[2]], key=lambda x: x[0].lower())

for name, size, is_dir, modify in dirs + files:
item = QTreeWidgetItem()
icon = "📁" if is_dir else "📄"
if name == "..":
icon = "⬆️"

item.setText(0, f"{icon} {name}")
item.setText(1, self._format_size(size) if not is_dir else "")
item.setText(2, "文件夹" if is_dir else "文件")
item.setData(0, Qt.ItemDataRole.UserRole, {
"name": name, "size": size, "is_dir": is_dir
})
self.tree.addTopLevelItem(item)

def _on_item_double_clicked(self, item, column):
data = item.data(0, Qt.ItemDataRole.UserRole)
if data:
self.file_selected.emit(data["name"], data["is_dir"])

def _show_context_menu(self, pos):
item = self.tree.itemAt(pos)
if not item:
return
data = item.data(0, Qt.ItemDataRole.UserRole)
if not data:
return

menu = QMenu(self)
if data["is_dir"]:
action = menu.addAction("打开目录")
action.triggered.connect(
lambda: self.file_selected.emit(data["name"], True)
)
if self.is_remote and data["name"] != "..":
menu.addSeparator()
del_action = menu.addAction("🗑️ 删除目录")
del_action.triggered.connect(
lambda: self.file_delete_requested.emit(data["name"], True)
)
else:
if self.is_remote:
action = menu.addAction("下载")
action.triggered.connect(
lambda: self.file_selected.emit(data["name"], False)
)
menu.addSeparator()
del_action = menu.addAction("删除文件")
del_action.triggered.connect(
lambda: self.file_delete_requested.emit(data["name"], False)
)
else:
action = menu.addAction("上传")
action.triggered.connect(
lambda: self.file_selected.emit(data["name"], False)
)
menu.exec(self.tree.viewport().mapToGlobal(pos))

def load_local_directory(self, path):
"""加载本地目录内容"""
if not os.path.isdir(path):
return
items = []
try:
if path != os.path.dirname(path): # 非根目录，添加 ..
items.append(("..", 0, True, ""))
for name in os.listdir(path):
full_path = os.path.join(path, name)
is_dir = os.path.isdir(full_path)
try:
size = os.path.getsize(full_path) if not is_dir else 0
except OSError:
size = 0
items.append((name, size, is_dir, ""))
self.set_file_list(items, path)
except PermissionError:
pass

@staticmethod
def _format_size(size_bytes):
if size_bytes < 1024:
return f"{size_bytes} B"
elif size_bytes < 1024 * 1024:
return f"{size_bytes / 1024:.1f} KB"
elif size_bytes < 1024 * 1024 * 1024:
return f"{size_bytes / (1024 * 1024):.1f} MB"
else:
return f"{size_bytes / (1024 * 1024 * 1024):.2f} GB"

# ============================================================
# 传输进度 Widget
# ============================================================
class TransferWidget(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
layout = QVBoxLayout(self)
layout.setContentsMargins(4, 4, 4, 4)

header = QLabel("<b>传输队列</b>")
layout.addWidget(header)

self.transfer_list = QTreeWidget()
self.transfer_list.setHeaderLabels(["文件", "方向", "进度", "状态"])
self.transfer_list.setColumnCount(4)
self.transfer_list.setRootIsDecorated(False)

header_view = self.transfer_list.header()
header_view.setStretchLastSection(True)
header_view.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
header_view.setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents)
header_view.setSectionResizeMode(3, QHeaderView.ResizeMode.ResizeToContents)

layout.addWidget(self.transfer_list)
self._progress_map = {}

def add_transfer(self, filename, direction):
"""direction: 'upload' or 'download'"""
item = QTreeWidgetItem()
icon = "📤" if direction == "upload" else "📥"
item.setText(0, filename)
item.setText(1, icon)
item.setText(2, "0%")
item.setText(3, "等待中")
self.transfer_list.addTopLevelItem(item)

progress = QProgressBar()
progress.setMaximum(100)
progress.setValue(0)
self.transfer_list.setItemWidget(item, 2, progress)
self._progress_map[filename] = (item, progress)

def update_progress(self, filename, transferred, total):
if filename in self._progress_map:
item, progress = self._progress_map[filename]
if total > 0:
pct = int(transferred * 100 / total)
progress.setValue(pct)
item.setText(2, f"{pct}%")
item.setText(3, "传输中…")

def complete_transfer(self, filename, success, message):
if filename in self._progress_map:
item, progress = self._progress_map[filename]
progress.setValue(100 if success else 0)
item.setText(2, "完成" if success else "失败")
item.setText(3, message)
if success:
item.setForeground(3, QColor("#2ecc71"))
else:
item.setForeground(3, QColor("#e74c3c"))

def clear_completed(self):
items_to_remove = []
for i in range(self.transfer_list.topLevelItemCount()):
item = self.transfer_list.topLevelItem(i)
status = item.text(3)
if status in ("上传成功", "下载成功", "上传失败", "下载失败"):
items_to_remove.append(i)
for i in reversed(items_to_remove):
self.transfer_list.takeTopLevelItem(i)

# ============================================================
# 调试控制台 Widget
# ============================================================
class DebugConsoleWidget(QWidget):
command_submitted = pyqtSignal(str) # 用户输入的原始命令

def __init__(self, parent=None):
super().__init__(parent)
layout = QVBoxLayout(self)
layout.setContentsMargins(4, 4, 4, 4)

# 标题行
header = QHBoxLayout()
title = QLabel("<b>调试控制台</b>")
header.addWidget(title)
header.addStretch()

clear_btn = QPushButton("清空")
clear_btn.setFixedWidth(60)
clear_btn.clicked.connect(self.clear_log)
header.addWidget(clear_btn)

save_btn = QPushButton("保存日志")
save_btn.setFixedWidth(80)
save_btn.clicked.connect(self.save_log)
header.addWidget(save_btn)

layout.addLayout(header)

# 日志显示区
self.log_text = QTextEdit()
self.log_text.setReadOnly(True)
self.log_text.setFont(QFont("Consolas", 9))
layout.addWidget(self.log_text)

# 命令输入行
cmd_layout = QHBoxLayout()
cmd_label = QLabel("FTP 命令:")
cmd_layout.addWidget(cmd_label)

self.cmd_input = QLineEdit()
self.cmd_input.setPlaceholderText("输入 FTP 命令，如: LIST /, SIZE file.txt, MKD newdir, PWD …")
self.cmd_input.setFont(QFont("Consolas", 10))
self.cmd_input.returnPressed.connect(self._on_send_command)
cmd_layout.addWidget(self.cmd_input)

send_btn = QPushButton("发送")
send_btn.setFixedWidth(60)
send_btn.clicked.connect(self._on_send_command)
cmd_layout.addWidget(send_btn)

layout.addLayout(cmd_layout)

# 命令历史
self._cmd_history = []
self._history_idx = –1

def append_log(self, category, message):
"""添加日志消息"""
timestamp = datetime.now().strftime("%H:%M:%S")

color_map = {
"cmd": "#e67e22", # 橙色 – 发送的命令
"resp": "#3498db", # 蓝色 – 服务器响应
"info": "#95a5a6", # 灰色 – 信息
"error": "#e74c3c", # 红色 – 错误
"success": "#2ecc71", # 绿色 – 成功
}

prefix_map = {
"cmd": ">>>",
"resp": "<<<",
"info": " ℹ️",
"error": " ❌",
"success": " ✅",
}

color = color_map.get(category, "#333333")
prefix = prefix_map.get(category, " ")

html = f'<span style="color:#888;">[{timestamp}]</span> '
html += f'<span style="color:{color};font-weight:bold;">{prefix}</span> '
html += f'<span style="color:{color};">{self._escape_html(message)}</span>'

self.log_text.append(html)
# 自动滚动到底部
scrollbar = self.log_text.verticalScrollBar()
scrollbar.setValue(scrollbar.maximum())

def clear_log(self):
self.log_text.clear()

def save_log(self):
path, _ = QFileDialog.getSaveFileName(
self, "保存日志", "ftp_debug_log.txt",
"文本文件 (*.txt);;所有文件 (*)"
)
if path:
with open(path, "w", encoding="utf-8") as f:
f.write(self.log_text.toPlainText())

def _on_send_command(self):
cmd = self.cmd_input.text().strip()
if cmd:
self._cmd_history.insert(0, cmd)
self._history_idx = –1
self.command_submitted.emit(cmd)
self.cmd_input.clear()

def keyPressEvent(self, event):
if event.key() == Qt.Key.Key_Up:
if self._cmd_history:
self._history_idx = min(
self._history_idx + 1, len(self._cmd_history) – 1
)
self.cmd_input.setText(self._cmd_history[self._history_idx])
elif event.key() == Qt.Key.Key_Down:
if self._cmd_history and self._history_idx > 0:
self._history_idx -= 1
self.cmd_input.setText(self._cmd_history[self._history_idx])
elif self._history_idx == 0:
self._history_idx = –1
self.cmd_input.clear()
else:
super().keyPressEvent(event)

@staticmethod
def _escape_html(text):
return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")

# ============================================================
# 主窗口
# ============================================================
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("FTP 客户端调试助手")
self.setMinimumSize(1100, 750)

# FTP 工作线程
self.ftp_worker = FtpWorkerThread()
self.ftp_worker.signals.log_message.connect(self._on_log_message)
self.ftp_worker.signals.file_list_received.connect(self._on_file_list_received)
self.ftp_worker.signals.transfer_progress.connect(self._on_transfer_progress)
self.ftp_worker.signals.transfer_complete.connect(self._on_transfer_complete)
self.ftp_worker.signals.connection_changed.connect(self._on_connection_changed)
self.ftp_worker.signals.raw_command_result.connect(self._on_raw_command_result)
self.ftp_worker.start()

self._init_ui()
self._load_settings()

def _init_ui(self):
central = QWidget()
self.setCentralWidget(central)
main_layout = QVBoxLayout(central)
main_layout.setContentsMargins(8, 8, 8, 8)
main_layout.setSpacing(6)

# —- 1. 连接管理 + 服务器控制 —-
conn_group = QGroupBox("连接管理")
conn_layout = QGridLayout(conn_group)
conn_layout.setContentsMargins(8, 12, 8, 8)
conn_layout.setHorizontalSpacing(8)
conn_layout.setVerticalSpacing(6)

# 状态指示灯
self.status_indicator = QLabel("●")
self.status_indicator.setStyleSheet("color: #e74c3c; font-size: 18px;")
self.status_indicator.setToolTip("未连接")
conn_layout.addWidget(self.status_indicator, 0, 0)

# 主机
conn_layout.addWidget(QLabel("主机:"), 0, 1)
self.host_input = QLineEdit("127.0.0.1")
self.host_input.setFixedWidth(150)
conn_layout.addWidget(self.host_input, 0, 2)

# 端口
conn_layout.addWidget(QLabel("端口:"), 0, 3)
self.port_input = QLineEdit("2121")
self.port_input.setFixedWidth(70)
conn_layout.addWidget(self.port_input, 0, 4)

# 用户名
conn_layout.addWidget(QLabel("用户名:"), 0, 5)
self.user_input = QLineEdit("admin")
self.user_input.setFixedWidth(100)
conn_layout.addWidget(self.user_input, 0, 6)

# 密码
conn_layout.addWidget(QLabel("密码:"), 0, 7)
self.pass_input = QLineEdit("123456")
self.pass_input.setEchoMode(QLineEdit.EchoMode.Password)
self.pass_input.setFixedWidth(100)
conn_layout.addWidget(self.pass_input, 0, 8)

# 连接/断开按钮
self.connect_btn = QPushButton("连接")
self.connect_btn.setFixedWidth(70)
self.connect_btn.setStyleSheet(
"QPushButton { background-color: #2ecc71; color: white; "
"font-weight: bold; border-radius: 4px; padding: 4px; }"
"QPushButton:hover { background-color: #27ae60; }"
)
self.connect_btn.clicked.connect(self._on_connect)
conn_layout.addWidget(self.connect_btn, 0, 9)

self.disconnect_btn = QPushButton("断开")
self.disconnect_btn.setFixedWidth(70)
self.disconnect_btn.setEnabled(False)
self.disconnect_btn.setStyleSheet(
"QPushButton { background-color: #e74c3c; color: white; "
"font-weight: bold; border-radius: 4px; padding: 4px; }"
"QPushButton:hover { background-color: #c0392b; }"
"QPushButton:disabled { background-color: #bdc3c7; }"
)
self.disconnect_btn.clicked.connect(self._on_disconnect)
conn_layout.addWidget(self.disconnect_btn, 0, 10)
conn_layout.setColumnStretch(11, 1)
main_layout.addWidget(conn_group)

# —- 2. 文件浏览器（双栏） —-
file_splitter = QSplitter(Qt.Orientation.Horizontal)

# 本地文件浏览器（左侧）
self.local_browser = FileBrowserWidget("本地文件", is_remote=False)
self.local_browser.file_selected.connect(self._on_local_file_action)
self.local_browser.load_local_directory(os.path.expanduser("~"))

local_btn_layout = QHBoxLayout()
browse_btn = QPushButton("📂 浏览…")
browse_btn.clicked.connect(self._browse_local)
local_btn_layout.addWidget(browse_btn)

refresh_local_btn = QPushButton("🔄 刷新")
refresh_local_btn.clicked.connect(
lambda: self.local_browser.load_local_directory(
self.local_browser.current_path
)
)
local_btn_layout.addWidget(refresh_local_btn)
local_btn_layout.addStretch()

local_container = QWidget()
local_container_layout = QVBoxLayout(local_container)
local_container_layout.setContentsMargins(0, 0, 0, 0)
local_container_layout.setSpacing(2)
local_container_layout.addWidget(self.local_browser)
local_container_layout.addLayout(local_btn_layout)

file_splitter.addWidget(local_container)

# 远程文件浏览器（右侧）
self.remote_browser = FileBrowserWidget("远程文件", is_remote=True)
self.remote_browser.file_selected.connect(self._on_remote_file_action)
self.remote_browser.file_delete_requested.connect(
self._on_remote_delete_request
)

# 上一级 + 刷新按钮
remote_btn_layout = QHBoxLayout()
up_remote_btn = QPushButton("⬆️ 上一级")
up_remote_btn.clicked.connect(self._go_up_remote)
remote_btn_layout.addWidget(up_remote_btn)

refresh_remote_btn = QPushButton("🔄 刷新")
refresh_remote_btn.clicked.connect(self._refresh_remote)
remote_btn_layout.addWidget(refresh_remote_btn)
remote_btn_layout.addStretch()

remote_container = QWidget()
remote_container_layout = QVBoxLayout(remote_container)
remote_container_layout.setContentsMargins(0, 0, 0, 0)
remote_container_layout.setSpacing(2)
remote_container_layout.addWidget(self.remote_browser)
remote_container_layout.addLayout(remote_btn_layout)

file_splitter.addWidget(remote_container)

file_splitter.setStretchFactor(0, 1)
file_splitter.setStretchFactor(1, 1)

main_layout.addWidget(file_splitter, stretch=3)

# —- 3. 传输队列 —-
self.transfer_widget = TransferWidget()
main_layout.addWidget(self.transfer_widget, stretch=1)

# —- 4. 调试控制台 —-
self.debug_console = DebugConsoleWidget()
# 将调试控制台的命令提交信号连接到工作线程
self.debug_console.command_submitted.connect(
self.ftp_worker.execute_raw_command
)
main_layout.addWidget(self.debug_console, stretch=2)

# —- 状态栏 —-
self.statusBar().showMessage("就绪")

# —- 连接操作 —-
def _on_connect(self):
host = self.host_input.text().strip()
port = self.port_input.text().strip()
user = self.user_input.text().strip()
password = self.pass_input.text().strip()

if not host or not port:
QMessageBox.warning(self, "提示", "请输入主机地址和端口")
return

self.connect_btn.setEnabled(False)
self.statusBar().showMessage(f"正在连接 {host}:{port}…")
self.ftp_worker.connect_to_server(host, port, user, password)

def _on_disconnect(self):
self.ftp_worker.disconnect_from_server()
self.remote_browser.tree.clear()
self.remote_browser.path_label.clear()

def _on_connection_changed(self, connected):
self.connect_btn.setEnabled(not connected)
self.disconnect_btn.setEnabled(connected)

if connected:
self.status_indicator.setStyleSheet(
"color: #2ecc71; font-size: 18px;"
)
self.status_indicator.setToolTip("已连接")
self.statusBar().showMessage("已连接")
else:
self.status_indicator.setStyleSheet(
"color: #e74c3c; font-size: 18px;"
)
self.status_indicator.setToolTip("未连接")
self.statusBar().showMessage("已断开")

# —- 文件浏览 —-
def _on_file_list_received(self, items, current_path):
self.remote_browser.set_file_list(items, current_path)
self.statusBar().showMessage(
f"远程目录: {current_path} ({len(items)} 个项目)"
)

def _on_remote_file_action(self, name, is_dir):
if is_dir:
if name == "..":
self.ftp_worker.change_directory("..")
else:
self.ftp_worker.change_directory(name)
else:
# 文件被选中 — 弹出下载对话框
self._prompt_download(name)

def _on_remote_delete_request(self, name, is_dir):
kind = "目录" if is_dir else "文件"
reply = QMessageBox.question(
self, "确认删除",
f"确定要删除远程{kind}吗？\\n\\n名称: {name}\\n\\n"
"注意: 此操作不可恢复。",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
if is_dir:
self.ftp_worker.delete_directory(name)
else:
self.ftp_worker.delete_file(name)

def _on_local_file_action(self, name, is_dir):
if is_dir:
if name == "..":
parent = os.path.dirname(self.local_browser.current_path)
self.local_browser.load_local_directory(parent)
else:
new_path = os.path.join(self.local_browser.current_path, name)
self.local_browser.load_local_directory(new_path)
else:
# 本地文件被选中 — 弹出上传对话框
full_path = os.path.join(self.local_browser.current_path, name)
self._prompt_upload(full_path)

def _go_up_remote(self):
self.ftp_worker.change_directory("..")

def _refresh_remote(self):
self.ftp_worker.list_directory()

def _browse_local(self):
directory = QFileDialog.getExistingDirectory(
self, "选择本地目录", self.local_browser.current_path
)
if directory:
self.local_browser.load_local_directory(directory)

# —- 文件传输 —-
def _prompt_download(self, remote_name):
save_path, _ = QFileDialog.getSaveFileName(
self, "下载文件", os.path.join(os.path.expanduser("~"), remote_name)
)
if save_path:
self.transfer_widget.add_transfer(remote_name, "download")
self.ftp_worker.download_file(remote_name, save_path)

def _prompt_upload(self, local_path):
remote_name = os.path.basename(local_path)
reply = QMessageBox.question(
self, "确认上传",
f"上传文件到远程当前目录？\\n\\n文件: {remote_name}",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
self.transfer_widget.add_transfer(remote_name, "upload")
self.ftp_worker.upload_file(local_path, remote_name)

def _on_transfer_progress(self, filename, transferred, total):
self.transfer_widget.update_progress(filename, transferred, total)

def _on_transfer_complete(self, filename, success, message):
self.transfer_widget.complete_transfer(filename, success, message)
# 传输完成后刷新远程目录
if success:
self.ftp_worker.list_directory()

# —- 日志 —-
def _on_log_message(self, category, message):
self.debug_console.append_log(category, message)
if category == "info":
self.statusBar().showMessage(message)

def _on_raw_command_result(self, result):
self.debug_console.append_log("resp", f"[命令结果] {result}")

# —- 设置保存/加载 —-
def _load_settings(self):
settings = QSettings("FtpDebugAssistant", "FTPHelper")
host = settings.value("host", "127.0.0.1")
port = settings.value("port", "2121")
user = settings.value("user", "admin")
self.host_input.setText(host)
self.port_input.setText(str(port))
self.user_input.setText(user)

def _save_settings(self):
settings = QSettings("FtpDebugAssistant", "FTPHelper")
settings.setValue("host", self.host_input.text())
settings.setValue("port", self.port_input.text())
settings.setValue("user", self.user_input.text())

def closeEvent(self, event):
self._save_settings()
self.ftp_worker.stop()
event.accept()

# ============================================================
# 入口
# ============================================================
if __name__ == "__main__":
app = QApplication(sys.argv)
app.setStyle("Fusion")

# 设置调色板
palette = QPalette()
palette.setColor(QPalette.ColorRole.Window, QColor(245, 245, 245))
palette.setColor(QPalette.ColorRole.WindowText, QColor(33, 33, 33))
palette.setColor(QPalette.ColorRole.Base, QColor(255, 255, 255))
palette.setColor(QPalette.ColorRole.AlternateBase, QColor(240, 240, 240))
palette.setColor(QPalette.ColorRole.ToolTipBase, QColor(255, 255, 255))
palette.setColor(QPalette.ColorRole.ToolTipText, QColor(33, 33, 33))
palette.setColor(QPalette.ColorRole.Text, QColor(33, 33, 33))
palette.setColor(QPalette.ColorRole.Button, QColor(240, 240, 240))
palette.setColor(QPalette.ColorRole.ButtonText, QColor(33, 33, 33))
palette.setColor(QPalette.ColorRole.Highlight, QColor(52, 152, 219))
palette.setColor(QPalette.ColorRole.HighlightedText, QColor(255, 255, 255))
app.setPalette(palette)

# 全局样式
app.setStyleSheet("""
QGroupBox {
font-weight: bold;
border: 1px solid #ddd;
border-radius: 6px;
margin-top: 10px;
padding-top: 14px;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 10px;
padding: 0 5px;
}
QLineEdit {
border: 1px solid #ccc;
border-radius: 4px;
padding: 3px 6px;
background: white;
}
QLineEdit:focus {
border-color: #3498db;
}
QPushButton {
border: 1px solid #ccc;
border-radius: 4px;
padding: 4px 12px;
background: #f0f0f0;
}
QPushButton:hover {
background: #e0e0e0;
}
QPushButton:pressed {
background: #d0d0d0;
}
QTreeWidget {
border: 1px solid #ddd;
border-radius: 4px;
background: white;
alternate-background-color: #f8f8f8;
}
QTreeWidget::item:selected {
background-color: #3498db;
color: white;
}
QHeaderView::section {
background-color: #ecf0f1;
padding: 4px;
border: 1px solid #ddd;
font-weight: bold;
}
QTextEdit {
border: 1px solid #ddd;
border-radius: 4px;
background: #1e1e1e;
color: #d4d4d4;
}
QProgressBar {
border: 1px solid #ddd;
border-radius: 4px;
text-align: center;
}
QProgressBar::chunk {
background-color: #3498db;
border-radius: 3px;
}
""")

window = MainWindow()
window.show()
sys.exit(app.exec())

4、ftp_server_test.py（测试使用）

"""
FTP 服务器测试例程
用于测试 FTP 客户端调试助手
"""

import os
import sys
import threading
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer

class SimpleFTPServer:
"""简单的 FTP 服务器"""

def __init__(self, host="127.0.0.1", port=2121, directory="./ftp_root",
username="admin", password="123456"):
self.host = host
self.port = port
self.directory = directory
self.username = username
self.password = password
self.server = None
self.running = False

# 确保目录存在
os.makedirs(directory, exist_ok=True)

# 创建一些测试文件
self._create_test_files()

def _create_test_files(self):
"""创建测试文件"""
test_files = [
("test1.txt", "这是测试文件 1"),
("test2.txt", "这是测试文件 2"),
("readme.txt", "FTP 服务器测试文件\\n用于测试客户端功能"),
]

for filename, content in test_files:
filepath = os.path.join(self.directory, filename)
if not os.path.exists(filepath):
with open(filepath, 'w', encoding='utf-8') as f:
f.write(content)
print(f"✓ 创建测试文件: {filename}")

# 创建子目录
subdir = os.path.join(self.directory, "subdir")
os.makedirs(subdir, exist_ok=True)
if not os.path.exists(os.path.join(subdir, "test3.txt")):
with open(os.path.join(subdir, "test3.txt"), 'w', encoding='utf-8') as f:
f.write("这是子目录中的测试文件")
print(f"✓ 创建子目录和文件: subdir/test3.txt")

def start(self):
"""启动 FTP 服务器"""
try:
# 创建授权器
authorizer = DummyAuthorizer()

# 添加用户（完整权限）
authorizer.add_user(
self.username,
self.password,
self.directory,
perm="elradfmw"
)

# 添加匿名用户（只读）
authorizer.add_anonymous(self.directory, perm="elr")

# 创建 FTP 处理器
handler = FTPHandler
handler.authorizer = authorizer

# 设置被动模式端口范围
handler.passive_ports = range(60000, 60100)

# 创建服务器
self.server = FTPServer((self.host, self.port), handler)
self.server.max_cons = 256
self.server.max_cons_per_ip = 5

self.running = True
print(f"\\n{'='*60}")
print(f"FTP 服务器已启动")
print(f"{'='*60}")
print(f"主机: {self.host}")
print(f"端口: {self.port}")
print(f"用户名: {self.username}")
print(f"密码: {self.password}")
print(f"根目录: {self.directory}")
print(f"{'='*60}")
print(f"\\n按 Ctrl+C 停止服务器\\n")

# 启动服务器
self.server.serve_forever()

except KeyboardInterrupt:
print("\\n\\n正在停止服务器…")
self.stop()
except Exception as e:
print(f"服务器错误: {e}")
self.running = False

def stop(self):
"""停止 FTP 服务器"""
if self.server:
self.server.close_all()
self.running = False
print("✓ 服务器已停止")

def main():
"""主函数"""
print("\\n" + "="*60)
print("FTP 服务器测试例程")
print("="*60)

# 默认配置
host = "127.0.0.1"
port = 2121
directory = r"D:\\user\\01417804\\桌面\\PythonProject\\image"
username = "admin"
password = "123456"

print("\\n服务器配置:")
print(f" 主机: {host} (回车使用默认值)")
print(f" 端口: {port} (回车使用默认值)")
print(f" 目录: {directory} (回车使用默认值)")
print(f" 用户名: {username} (回车使用默认值)")
print(f" 密码: {password} (回车使用默认值)")
print()

# 获取用户输入
host = input("主机地址: ").strip() or host
port = int(input("端口号: ").strip() or port)
directory = input("根目录: ").strip() or directory
username = input("用户名: ").strip() or username
password = input("密码: ").strip() or password

# 创建并启动服务器
server = SimpleFTPServer(host, port, directory, username, password)
server.start()

if __name__ == "__main__":
main()

![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/78c60869b1d04588b9f845e062311394

三、实现原理

本工具的核心设计思路是：在 ftplib.FTP 的基础上做一层“协议级”的拦截与封装，配合 Qt 的信号槽机制，把耗时的网络操作放到后台线程执行，从而既保证界面流畅，又能把每一次 FTP 命令与响应实时呈现到调试控制台。 下面从几个关键点展开说明。

1、协议拦截：继承 FTP 重写 putline / getline

FTP 协议本质上是基于 TCP 的文本协议，客户端与服务器之间通过 ASCII 命令和响应进行交互。ftplib.FTP 内部把“发送命令”和“读取响应”分别封装在 putline() 与 getline() 两个方法中。

因此，只要继承 FTP 并重写这两个方法，就能在不改动任何业务逻辑的前提下，捕获到客户端发出的每一条命令和服务器返回的每一条响应：

class DebugFTP(FTP):
def putline(self, line):
if self._log_callback:
self._log_callback("cmd", line) # 拦截发出的命令
super().putline(line)

def getline(self):
line = super().getline()
if self._log_callback:
self._log_callback("resp", line) # 拦截收到的响应
return line

• putline 在命令真正写入 socket 之前被调用，此时 line 就是即将发送的原始命令文本（如 USER admin、PWD、STOR test.txt）。
• getline 在从 socket 读取到一行响应后被调用，此时 line 就是服务器返回的原始响应（如 230 Login successful.、257 "/" is the current directory.）。
• 通过构造函数传入的 log_callback 回调，把 ("cmd", line) 或 ("resp", line) 交给上层，再由信号 log_message 转发到调试控制台。

这样做的最大好处是零侵入：ftplib 内部所有命令（login、cwd、storbinary、retrbinary、mlsd 等）最终都会走这两个方法，因此日志覆盖是完整且自动的，无需为每个命令单独埋点。

2、线程模型：QThread + 任务队列

FTP 网络操作（连接、列目录、上传、下载）都是阻塞调用，如果直接放在 GUI 主线程执行，界面会卡死。因此本项目把 FtpWorkerThread 作为唯一的 FTP 操作执行者，运行在后台线程中。

2.1 任务队列

工作线程内部维护一个线程安全的任务队列：

self._task_queue = []
self._lock = threading.Lock()

def _enqueue(self, task_type, params):
with self._lock:
self._task_queue.append((task_type, params))

主线程（界面）通过 connect_to_server()、upload_file() 等方法向队列追加任务，工作线程在 run() 中循环取出并执行：

def run(self):
while not self._stop_flag:
task = None
with self._lock:
if self._task_queue:
task = self._task_queue.pop(0)
if task is None:
self.msleep(50)
continue
task_type, params = task
try:
# 根据 task_type 分发到对应的 _do_xxx 方法
...
except Exception as e:
self.signals.log_message.emit("error", f"操作失败: {str(e)}")

• 队列为空时线程 msleep(50) 让出 CPU，避免空转。
• 所有任务在同一个后台线程中串行执行，天然避免了多个 FTP 操作并发导致的协议状态错乱（FTP 是有状态协议，同一连接上并发操作会互相干扰）。
• 每个任务用 try/except 包裹，任何异常都不会导致线程退出，而是通过信号上报给界面。

2.2 线程安全

• 任务入队使用 threading.Lock 保护，保证主线程与工作线程对队列的并发访问安全。
• 工作线程不直接操作任何界面控件，只通过 FtpSignals 信号把结果发回主线程；主线程中的槽函数（如 _on_file_list_received）负责更新界面。这符合 Qt 的线程规则：只有主线程可以操作 GUI 控件。

3、信号槽通信：FtpSignals

FtpSignals 继承自 QObject，集中定义了工作线程与主线程之间的所有通信信号：

class FtpSignals(QObject):
log_message = pyqtSignal(str, str) # (类别, 消息)
file_list_received = pyqtSignal(list, str) # (文件列表, 当前路径)
transfer_progress = pyqtSignal(str, int, int) # (文件名, 已传输, 总大小)
transfer_complete = pyqtSignal(str, bool, str) # (文件名, 成功, 消息)
connection_changed = pyqtSignal(bool) # 连接状态
raw_command_result = pyqtSignal(str) # 原始命令执行结果

• log_message：携带类别（cmd / resp / info / error / success）和消息文本，调试控制台据此用不同颜色渲染。
• file_list_received：携带解析后的文件列表和当前路径，远程文件浏览器据此刷新。
• transfer_progress / transfer_complete：驱动传输队列的进度条和完成状态。
• connection_changed：通知主线程更新连接按钮的可用状态和状态指示灯。

由于 pyqtSignal 是跨线程安全的，工作线程 emit 的信号会自动排队到主线程的事件循环中执行对应的槽函数，从而实现了线程间安全通信。

4、目录列表解析：MLSD 优先，LIST 兜底

FTP 列目录有两种常见方式：

• MLSD：返回结构化数据（类型、大小、修改时间等），解析简单可靠，但需要服务器支持（RFC 3659）。
• LIST：返回人类可读的文本列表，格式因服务器而异（Unix 风格、DOS 风格等），解析较繁琐。

本工具的策略是优先尝试 MLSD，失败则回退到 LIST：

try:
for name, facts in self.ftp.mlsd():
is_dir = facts.get("type", "") == "dir"
size = int(facts.get("size", 0))
modify = facts.get("modify", "")
items.append((name, size, is_dir, modify))
except error_perm:
lines = []
self.ftp.retrlines("LIST", lines.append)
for line in lines:
parsed = self._parse_list_line(line)
if parsed:
items.append(parsed)

_parse_list_line 同时兼容 Unix 和 DOS 两种常见格式：

• Unix 格式：drwxr-xr-x 2 user group 4096 Jan 01 12:00 filename，通过权限位首字符判断类型，取第 5 个字段为大小，第 9 个字段为文件名。
• DOS 格式：01-01-24 12:00AM <DIR> dirname，通过 <DIR> 标记判断目录，否则取数字字段为大小。

5、上传/下载进度跟踪

5.1 上传：包装文件对象

上传时通过 storbinary 逐块读取本地文件并发送。为了跟踪进度，把文件对象包装一层，在每次 read() 时累计已读字节数：

class TrackingFile:
def read(self, *args, **kwargs):
data = self._f.read(*args, **kwargs)
if data:
transferred[0] += len(data)
now = time.time()
if now – last_emit[0] > 0.1:
self.signals.transfer_progress.emit(filename, transferred[0], file_size)
last_emit[0] = now
return data

• 用 transferred[0] 列表而非普通变量，是为了在闭包中修改外部变量。
• 通过 time.time() 做节流：每 100ms 才发一次进度信号，避免高频信号刷爆界面。

5.2 下载：回调函数

下载时通过 retrbinary 的 callback 参数接收数据块，同样累计字节数并节流发送进度：

def on_data(data):
f.write(data)
transferred[0] += len(data)
now = time.time()
if now – last_emit[0] > 0.1:
self.signals.transfer_progress.emit(filename, transferred[0], file_size)
last_emit[0] = now

下载前先通过 ftp.size() 获取文件总大小，用于计算百分比；若获取失败则进度条显示为不确定状态。

6、原始命令执行

调试控制台允许用户手动输入任意 FTP 命令。_do_raw_command 对常见命令做了专门处理：

• LIST / NLST：通过 retrlines 获取多行结果。
• PWD / CWD / SIZE / MKD / RMD / DELE / REN：调用 ftplib 对应方法。
• 其他命令：通过 sendcmd 原样发送并返回服务器响应。

else:
resp = self.ftp.sendcmd(cmd)
result = resp

这样既覆盖了常用操作，又保留了任意命令透传的能力，方便调试非标准或扩展的 FTP 命令。

7、整体数据流

#mermaid-svg-BFB4g9mWoCDucUl4{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-BFB4g9mWoCDucUl4 .error-icon{fill:#552222;}#mermaid-svg-BFB4g9mWoCDucUl4 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-BFB4g9mWoCDucUl4 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-BFB4g9mWoCDucUl4 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-BFB4g9mWoCDucUl4 .marker.cross{stroke:#333333;}#mermaid-svg-BFB4g9mWoCDucUl4 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-BFB4g9mWoCDucUl4 p{margin:0;}#mermaid-svg-BFB4g9mWoCDucUl4 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-BFB4g9mWoCDucUl4 .cluster-label text{fill:#333;}#mermaid-svg-BFB4g9mWoCDucUl4 .cluster-label span{color:#333;}#mermaid-svg-BFB4g9mWoCDucUl4 .cluster-label span p{background-color:transparent;}#mermaid-svg-BFB4g9mWoCDucUl4 .label text,#mermaid-svg-BFB4g9mWoCDucUl4 span{fill:#333;color:#333;}#mermaid-svg-BFB4g9mWoCDucUl4 .node rect,#mermaid-svg-BFB4g9mWoCDucUl4 .node circle,#mermaid-svg-BFB4g9mWoCDucUl4 .node ellipse,#mermaid-svg-BFB4g9mWoCDucUl4 .node polygon,#mermaid-svg-BFB4g9mWoCDucUl4 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-BFB4g9mWoCDucUl4 .rough-node .label text,#mermaid-svg-BFB4g9mWoCDucUl4 .node .label text,#mermaid-svg-BFB4g9mWoCDucUl4 .image-shape .label,#mermaid-svg-BFB4g9mWoCDucUl4 .icon-shape .label{text-anchor:middle;}#mermaid-svg-BFB4g9mWoCDucUl4 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-BFB4g9mWoCDucUl4 .rough-node .label,#mermaid-svg-BFB4g9mWoCDucUl4 .node .label,#mermaid-svg-BFB4g9mWoCDucUl4 .image-shape .label,#mermaid-svg-BFB4g9mWoCDucUl4 .icon-shape .label{text-align:center;}#mermaid-svg-BFB4g9mWoCDucUl4 .node.clickable{cursor:pointer;}#mermaid-svg-BFB4g9mWoCDucUl4 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-BFB4g9mWoCDucUl4 .arrowheadPath{fill:#333333;}#mermaid-svg-BFB4g9mWoCDucUl4 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-BFB4g9mWoCDucUl4 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-BFB4g9mWoCDucUl4 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-BFB4g9mWoCDucUl4 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-BFB4g9mWoCDucUl4 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-BFB4g9mWoCDucUl4 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-BFB4g9mWoCDucUl4 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-BFB4g9mWoCDucUl4 .cluster text{fill:#333;}#mermaid-svg-BFB4g9mWoCDucUl4 .cluster span{color:#333;}#mermaid-svg-BFB4g9mWoCDucUl4 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-BFB4g9mWoCDucUl4 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-BFB4g9mWoCDucUl4 rect.text{fill:none;stroke-width:0;}#mermaid-svg-BFB4g9mWoCDucUl4 .icon-shape,#mermaid-svg-BFB4g9mWoCDucUl4 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-BFB4g9mWoCDucUl4 .icon-shape p,#mermaid-svg-BFB4g9mWoCDucUl4 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-BFB4g9mWoCDucUl4 .icon-shape .label rect,#mermaid-svg-BFB4g9mWoCDucUl4 .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-BFB4g9mWoCDucUl4 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-BFB4g9mWoCDucUl4 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-BFB4g9mWoCDucUl4 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

用户点击界面按钮

主线程调用 FtpWorkerThread 的入队方法

任务写入线程安全队列

后台线程 run() 取出任务

执行 FTP 操作（DebugFTP）

putline/getline 拦截命令与响应

通过 FtpSignals 信号发回主线程

主线程槽函数更新界面

调试控制台 / 文件浏览器 / 进度条

8、小结

通过以上设计，本工具既是一个可用的 FTP 客户端，更是一个协议级调试器——所有交互过程完全透明可见，非常适合学习 FTP 协议或排查服务器问题。

er)