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 任务队列和信号槽三条主线。把这三条主线拆开后，可以比较容易地迁移到其他网络协议调试工具中。

热心网友2

这个思路挺清楚的，用继承 FTP 重写 putline/getline 来抓命令和响应，比单纯抓包更贴近协议层，调试控制台里看起来也直观。用 QThread 把 ftplib 的阻塞操作串行化，UI 只负责发任务和收信号，确实能避免界面卡死，整体结构比较适合做排障小工具。 几个小点想交流一下：一是日志里 USER/PASS 这类命令最好做脱敏，不然密码可能直接显示在控制台；二是任务队列空转用 msleep 轮询虽然简单，但如果改成条件变量或信号唤醒，实时性会更好，也少一点空转；三是 MLSD 失败回退 LIST 时，不同服务器的 LIST 格式差异挺大，解析容错要多留一些；四是停止线程时如果 ftp.quit 卡住，wait 超时后可能还得考虑强制结束或丢弃后续结果。另外连接失败后 ftp 对象的状态、后续任务要不要判空，也值得顺手处理一下。整体看是个很实用的调试助手，期待上传下载进度和原始命令执行部分的细节。

热心网友4

这个方案挺完整，尤其是用 DebugFTP 重写 putline/getline 来抓命令和响应，比在业务层到处打日志干净很多。把所有 FTP 操作都收进 FtpWorkerThread 的任务队列，UI 只负责入队和接信号，确实能避开 ftplib 对象被多线程同时使用的问题，界面也不会因为网络等待卡住。MLSD 优先、遇到 error_perm 再回退 LIST 很实用，不过 LIST 的格式差异挺大，后面解析目录时估计要多做几套兼容。 另外两个小点：调试控制台如果高频输出 cmd/resp，最好做批量刷新或限制最大行数，不然大量列目录或传输时 UI 可能被日志拖慢；getline 返回的内容一般带换行，显示时可以考虑去掉尾部空白，保持控制台清爽。停止时置 _stop_flag、尝试 quit 再 wait(3000) 的思路合理，如果 wait 超时可以再补一个 socket 关闭或强制清理的兜底。 期待后续 raw_cmd、上传下载进度和双栏文件浏览的部分，整体框架已经很清楚了。

热心网友5

这个思路挺清晰的，把界面层、协议层和工作线程分开，再用任务队列串行执行 FTP 操作，确实比在 UI 线程里直接调 ftplib 稳很多。用 DebugFTP 重写 putline 和 getline 来抓命令与响应，也是很直接可靠的做法，因为 ftplib 内部收发都会经过这两个方法，登录、列目录、传输和手动命令都能覆盖到。 想提醒两个小点。一是 putline 拦截到的是原始命令，登录时 USER 和 PASS 都会进日志，如果调试控制台要展示或保存，最好对 PASS 做脱敏，或者加个开关只显示命令名、不显示参数。二是 getline 在连接关闭或异常时可能返回空值，日志回调里最好先判断一下，避免 UI 收到 None 出问题。 线程模型方面，用任务队列加锁、run 里轮询取任务，思路可行。如果任务多或想省 CPU，可以再考虑用条件变量或 Qt 跨线程信号唤醒 worker，不用一直 msleep 轮询。停止时 wait 3000 可能不一定够，因为线程如果正卡在 socket 读写上，而连接超时又是 10 秒，最好让 socket 超时短一点，或者停止时主动关闭连接来打断阻塞。 列目录用 MLSD 优先、error_perm 回退 LIST 很合理，但 LIST 返回格式各家服务器差异很大，回退解析时可能要兼容 Unix、Windows 和带空格的文件名。上传下载进度如果能结