PyQt6实现FTP调试助手：命令拦截与QThread任务队列

脚本专家

在需要查看 FTP 协议细节、手动发送原始命令或排查上传下载问题时，普通 FTP 客户端往往只给出结果，不展示命令/响应过程。原文给出的方案是用 PyQt6 做界面，用 ftplib 做协议层，并通过继承 FTP 拦截 putline/getline，把每一条命令和响应送到调试控制台；后台用 QThread 任务队列串行执行 FTP 操作，避免 UI 卡死。

功能包括：连接/断开 FTP 服务器、本地/远程双栏文件浏览器、带进度的上传/下载、调试控制台实时日志与手动原始命令。工程结构中有 main.py 和 ftp_server_test.py 测试脚本，依赖 PyQt6；FTP 部分来自 Python 标准库 ftplib。

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

2. class DebugFTP(FTP):
3.     def __init__(self, *args, log_callback=None, **kwargs):
4.         self._log_callback = log_callback
5.         super().__init__(*args, **kwargs)
7.     def putline(self, line):
8.         if self._log_callback:
9.             self._log_callback('cmd', line)
10.         super().putline(line)
12.     def getline(self):
13.         line = super().getline()
14.         if self._log_callback:
15.             self._log_callback('resp', line)
16.         return line

复制代码

这里 log_callback 接收类别和消息，类别为 cmd/resp。FtpWorkerThread 连接时用 DebugFTP(log_callback=self._ftp_log) 构造客户端，之后所有协议交互都能进入日志回调，再由信号槽转发到 UI。

二、线程模型与任务队列
所有 FTP 操作放在 FtpWorkerThread(QThread) 中，run 循环从 _task_queue 取任务。队列由 threading.Lock 保护，任务类型包括 connect、disconnect、list、cwd、upload、download、raw_cmd、delete_file、delete_dir。外部调用 connect_to_server、list_directory、upload_file 等方法时，只是把任务参数入队，不直接操作 FTP 对象。

2. def _enqueue(self, task_type, params):
3.     with self._lock:
4.         self._task_queue.append((task_type, params))
6. def run(self):
7.     while not self._stop_flag:
8.         task = None
9.         with self._lock:
10.             if self._task_queue:
11.                 task = self._task_queue.pop(0)
12.         if task is None:
13.             self.msleep(50)
14.             continue
15.         task_type, params = task
16.         try:
17.             if task_type == 'connect':
18.                 self._do_connect(**params)
19.             elif task_type == 'list':
20.                 self._do_list(**params)
21.             # 其余任务类型同理分发
22.         except Exception as e:
23.             self.signals.log_message.emit('error', f'操作失败: {str(e)}')

复制代码

这种模型让 UI 线程只负责收集输入和展示信号，FTP 网络等待不会阻塞界面。停止时把 _stop_flag 置为 True，若已连接则尝试 ftp.quit()，最后 wait(3000) 等待线程退出。

三、信号槽通信
UI 与工作线程之间通过 FtpSignals(QObject) 通信，信号包括：log_message(str, str)，类别 cmd/resp/info/error/success；file_list_received(list, str)，文件列表和当前路径；transfer_progress(str, int, int)，文件名、已传输、总大小；transfer_complete(str, bool, str)，文件名、是否成功、消息；connection_changed(bool)，连接状态；raw_command_result(str)，原始命令执行结果。线程中只 emit 信号，不直接改控件，这是 PyQt 多线程更新的常见做法。

四、连接与目录列表
_do_connect 中先用 DebugFTP 创建客户端，再 connect(host, int(port), timeout=10)、login(user, password)。连接成功后 emit connection_changed(True)，并立即调用 _do_list(path='') 列出根目录。列目录优先使用 MLSD：

2. try:
3.     for name, facts in self.ftp.mlsd():
4.         if name == '.':
5.             continue
6.         if name == '..':
7.             items.append(('..', 0, True, ''))
8.             continue
9.         is_dir = facts.get('type', '') == 'dir'
10.         size = int(facts.get('size', 0))
11.         modify = facts.get('modify', '')
12.         items.append((name, size, is_dir, modify))
13. except error_perm:
14.     lines = []
15.     self.ftp.retrlines('LIST', lines.append)
16.     for line in lines:
17.         parsed = self._parse_list_line(line)
18.         if parsed:
19.             items.append(parsed)
20. self.signals.file_list_received.emit(items, cwd)

复制代码

如果服务器不支持 MLSD，会抛出 error_perm，代码回退到 LIST，并通过 _parse_list_line 解析 Unix 与 DOS 两种常见格式。Unix 格式判断首字符是否为 d，并取 parts[4] 为大小、parts[8] 为文件名；DOS 格式判断 <DIR>，否则读取大小。这个兼容逻辑是该调试助手能连接不同 FTP 服务端的关键。

五、上传/下载进度
上传时用 TrackingFile 包装本地文件对象，在 read 时累计 transferred[0]，并以 0.1 秒为间隔节流 emit transfer_progress，避免进度信号过密；然后调用 storbinary('STOR {remote_name}', TrackingFile(f))。下载时先尝试 ftp.size(remote_name) 获取总大小，再通过 retrbinary('RETR {remote_name}', on_data)。on_data 负责写入本地文件、累计传输量，并同样按 0.1 秒节流发送进度；完成后 emit transfer_complete 和日志。

六、原始命令控制台
调试控制台除了自动记录协议日志，还支持手动输入原始命令。execute_raw_command 入队 raw_cmd 任务，_do_raw_command 先 split(None, 1) 解析命令名和参数，再针对常见命令做特殊处理：LIST、NLST、PWD、CWD、SIZE、MKD、RMD、DELE、REN、HELP；QUIT/BYE 会提示使用界面断开按钮；其他命令走 ftp.sendcmd(cmd)。执行结果通过 raw_command_result 返回，异常时返回“错误: ...”。这样既能方便地验证自定义 FTP 扩展命令，也能在排查服务器响应时保留完整上下文。

七、适用场景与注意事项
该方案适合做 FTP 协议调试、自动化脚本验证、上传下载问题定位。需要注意：MLSD 并非所有 FTP 都支持，代码已经用 error_perm 回退到 LIST；LIST 解析受服务器返回格式影响，Unix/DOS 之外的格式可能需要扩展；FTP 操作必须保持在工作线程，UI 只能通过 pyqtSignal 更新；连接超时为 10 秒，停止线程时最多等待 3000 毫秒。原文的 ftp_server_test.py 是测试脚本，可用于本地验证，但正文未展开其内容，实际使用时可以配合本地 FTP 服务器测试连接、列目录、上传下载和原始命令流程。

综上所述，这个 PyQt6+FTP 调试助手的核心不是界面，而是 DebugFTP 协议拦截、QThread 任务队列和信号槽三条主线。把这三条主线拆开后，可以比较容易地迁移到其他网络协议调试工具中。