3个坑避掉：手写实现ftp下载工具，告别API变更噩梦

刚把老项目的FTP模块升级到最新库，一跑直接崩了。日志里满屏 AttributeError: module 'ftplib' has no attribute 'listfiles'，代码里明明没动过调用逻辑。这种版本升级后 API 全变了的情况，在运维和后端开发里太常见了。依赖第三方库就像把命门交给别人，今天咱们不装现成的包，直接手写实现一个稳定的 FTP 下载工具，彻底解决这个痛点。

项目目标与痛点分析

做开发最怕的不是功能复杂，而是基础组件的不稳定。很多团队习惯直接 pip install 一个 FTP 库，或者用 Node.js 的 npm install 装个客户端。但现实是，PyPI 官方包 ftplib 虽然是标准库，但在某些跨平台环境或高并发场景下，其 API 行为并不总是一致。更糟糕的是，很多第三方封装库（如某些流行的 NPM/PyPI 官方包之外的非标准库）经常在大版本更新时改变接口签名，导致你的业务代码被迫重构。

我们要实现的目标很简单：

零依赖核心逻辑：基于 Python 标准库 ftplib 和 os 模块，不引入任何第三方网络库。
断点续传：支持大文件下载中断后继续，避免重复传输。
健壮性：自动处理连接超时、文件不存在、权限不足等常见异常。
可复现：代码结构清晰，方便转岗从业者直接复用到生产环境。

为什么强调手写实现？因为当你完全理解 FTP 协议（RFC 959）的数据传输通道建立过程时，你就不会被任何库的 Bug 或 API 变更所绑架。对于追求薪资竞争力的后端工程师来说，这种底层掌控力是面试加分项，也是解决线上疑难杂症的底气。

目录结构与工程化设计

为了避免代码一团糟，我们采用标准的 Python 项目结构。虽然这是一个单文件工具，但工程化思维要求我们将配置、核心逻辑、异常处理分离。

ftp_downloader/
├── config.py          # 存储 FTP 连接参数、超时时间等配置
├── core.py            # 核心下载逻辑，包含 FTP 连接管理与文件传输
├── utils.py           # 工具函数，如日志记录、文件大小格式化
├── exceptions.py      # 自定义异常类，便于上层捕获
├── main.py            # 入口文件，解析命令行参数并调用核心模块
└── requirements.txt   # 虽然核心无依赖，但可预留日志库等可选依赖

这种结构的好处在于，当未来需要扩展为多线程下载或支持 SFTP 时，只需修改 core.py 中的传输策略，而不必触碰 main.py 的业务逻辑。对于从其他语言（如 Java 或 Go）转岗的开发者来说，这种模块化的组织方式能显著降低理解成本，符合主流后端框架的设计范式。

核心代码实现与逐行讲解

1. 自定义异常与配置

首先定义专属异常，避免混淆标准的 ConnectionError。

# exceptions.py
class FTPDownloadError(Exception):"""FTP 下载过程中的通用错误"""passclass FileNotFound(FTPDownloadError):"""远程文件不存在"""pass

配置模块使用数据类（dataclass）保证类型安全：

# config.py
from dataclasses import dataclass@dataclass
class FTPConfig:host: strport: intuser: strpassword: strtimeout: int = 30remote_path: str = "/"

2. 核心下载逻辑

这是最关键的部分。我们需要处理 FTP 的被动模式（PASV）连接，并实现分块下载。

# core.py
import os
import ftplib
import time
from .config import FTPConfig
from .exceptions import FileNotFound, FTPDownloadError
from .utils import format_size, get_loggerlogger = get_logger("ftp_downloader")class FTPDownloader:def __init__(self, config: FTPConfig):self.config = configself.ftp = Nonedef connect(self):"""建立 FTP 连接，使用被动模式避免 NAT 穿透问题"""try:self.ftp = ftplib.FTP()self.ftp.connect(self.config.host, self.config.port, timeout=self.config.timeout)self.ftp.login(self.config.user, self.config.password)# 关键：设置被动模式，确保数据通道能正常建立self.ftp.set_pasv(True)logger.info(f"Connected to {self.config.host}")except ftplib.all_errors as e:raise FTPDownloadError(f"Connection failed: {e}")def get_file_size(self, remote_file: str) -> int:"""获取远程文件大小，用于进度计算和断点判断"""try:# SIZE 命令是 FTP 标准扩展，大多数服务器支持size = self.ftp.size(remote_file)if size is None:raise FTPDownloadError(f"Cannot determine size of {remote_file}")return sizeexcept ftplib.error_perm:raise FileNotFound(f"Remote file {remote_file} not found")def download(self, remote_file: str, local_file: str, resume: bool = False):"""下载文件，支持断点续传:param remote_file: 远程文件路径:param local_file: 本地保存路径:param resume: 是否启用断点续传"""self.connect()try:remote_size = self.get_file_size(remote_file)local_size = 0# 断点续传逻辑：检查本地已下载大小if resume and os.path.exists(local_file):local_size = os.path.getsize(local_file)if local_size >= remote_size:logger.info("File already complete.")returnlogger.info(f"Resuming from byte {local_size}")else:# 如果不是续传或文件不存在，从头开始if os.path.exists(local_file):os.remove(local_file)local_size = 0# 打开本地文件，模式取决于是否续传mode = 'ab' if resume and local_size > 0 else 'wb'with open(local_file, mode) as f:# 如果续传，需要 seek 到指定位置if local_size > 0:self.ftp.sendcmd(f"REST {local_size}")# 使用 retrievebinaryfile 进行二进制传输# blocksize 设为 8192，平衡内存占用与传输效率downloaded = 0start_time = time.time()def callback(chunk):nonlocal downloadeddownloaded += len(chunk)progress = (downloaded / remote_size) * 100# 每 5% 打印一次进度，避免日志爆炸if int(progress) % 5 == 0:speed = format_size(downloaded / (time.time() - start_time)) + "/s"logger.info(f"Progress: {progress:.1f}% | Speed: {speed}")self.ftp.retrbinary(f"RETR {remote_file}", callback, blocksize=8192)logger.info(f"Download completed: {local_file}")finally:# 确保资源释放，无论是否发生异常if self.ftp:self.ftp.quit()self.ftp = None

代码逐行解析重点：

self.ftp.set_pasv(True)：这是新手最容易踩的坑。在 NAT 环境或云服务器上，主动模式（PORT）往往因防火墙拦截而失败。强制被动模式是生产环境的标配。
self.ftp.size(remote_file)：注意，ftplib 的 size 方法依赖于服务器的 SIZE 命令支持。如果服务器较老不支持，会返回 None，必须做判空处理，否则后续进度计算会除以零。
REST 命令：断点续传的核心。REST 告诉服务器从哪个字节偏移量开始发送数据。这比在客户端读取文件再拼接要高效得多。
retrbinaryfile 的回调：不要在回调中做耗时操作（如写入数据库），只更新内存变量。进度打印也要节流，否则日志文件会迅速膨胀。

3. 工具函数

# utils.py
import logging
import sysdef get_logger(name):logger = logging.getLogger(name)if not logger.handlers:handler = logging.StreamHandler(sys.stdout)formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')handler.setFormatter(formatter)logger.addHandler(handler)logger.setLevel(logging.INFO)return loggerdef format_size(size: float) -> str:"""将字节数格式化为人类可读的大小"""for unit in ['B', 'KB', 'MB', 'GB']:if size < 1024.0:return f"{size:.2f} {unit}"size /= 1024.0return f"{size:.2f} TB"

运行与测试

1. 入口文件设计

使用 argparse 处理命令行参数，使工具具备 CLI 特性，方便集成到 Shell 脚本中。

# main.py
import argparse
from .core import FTPDownloader
from .config import FTPConfig
from .exceptions import FTPDownloadErrordef main():parser = argparse.ArgumentParser(description="Robust FTP Downloader")parser.add_argument("--host", required=True, help="FTP Server Host")parser.add_argument("--user", required=True, help="FTP User")parser.add_argument("--password", required=True, help="FTP Password")parser.add_argument("--remote", required=True, help="Remote File Path")parser.add_argument("--local", required=True, help="Local Save Path")parser.add_argument("--resume", action="store_true", help="Enable Resume")args = parser.parse_args()config = FTPConfig(host=args.host,port=21, # 默认端口user=args.user,password=args.password,remote_path=args.remote)try:downloader = FTPDownloader(config)downloader.download(args.remote, args.local, resume=args.resume)except FTPDownloadError as e:print(f"Error: {e}")exit(1)if __name__ == "__main__":main()

2. 测试场景覆盖

转岗开发者在接手项目时，必须验证边界情况。以下是必须测试的场景：

测试场景	预期结果	验证方法
文件不存在	抛出 `FileNotFound`	传入错误的远程路径
网络中断	抛出连接超时异常	在下载过程中拔网线或停止服务
断点续传	从上次中断处继续	手动中断下载，再次运行带 `--resume`
大文件下载	进度条正常，内存稳定	下载 1GB 以上文件，监控内存占用
权限不足	抛出权限错误	使用只读账号尝试上传或删除

在实际测试中，建议使用 ftp-server 搭建本地测试环境，避免依赖生产服务器。可以使用 Docker 快速启动一个 FTP 容器：

docker run -d --name test-ftp -p 21:21 fauria/ftp-server

优化扩展与避坑指南

1. 性能优化：多线程分片下载

对于超大文件（如 10GB 以上），单线程受限于带宽和磁盘 IO，速度可能不理想。优化方案是参考 NPM/PyPI 官方包中一些高性能下载器的思路，将文件分片，多线程并行下载。

实现思路：

获取文件总大小。
将文件划分为 N 个分片（如 10 个）。
每个线程使用 REST 命令定位到自己负责的字节区间。
每个线程下载到临时文件。
主线程合并临时文件。

注意：FTP 协议本身对并发连接有限制，过多线程可能导致连接被服务器拒绝。建议线程数控制在 4-8 之间，并根据服务器响应动态调整。

2. 安全性加固

密码脱敏：日志中严禁打印明文密码。上述代码中已规避，但需确保 print 或 logger 调用处不泄露敏感信息。
输入校验：对 remote_file 和 local_file 进行路径遍历攻击（Path Traversal）检查，防止恶意用户通过 ../../etc/passwd 等方式读取服务器敏感文件。

import os
def safe_join(base, target):base = os.path.abspath(base)target = os.path.abspath(target)if not target.startswith(base):raise ValueError("Invalid path")return os.path.join(base, target)

3. 常见避坑清单

编码问题：FTP 文件名可能包含非 ASCII 字符。ftplib 默认使用 ISO-8859-1 编码。如果文件名是中文，可能出现乱码。需在连接后设置 self.ftp.encoding = 'utf-8'（如果服务器支持）。
超时处理：ftplib 的 timeout 参数仅适用于控制通道。数据传输通道可能因网络波动卡住。建议结合 socket 层的超时机制，或设置最大下载时间。
资源泄漏：务必在 finally 块中关闭连接。如果在下载过程中发生异常而未关闭，会导致 FTP 服务器连接数耗尽，最终拒绝新连接。