Python自动化脚本:从网页批量提取FTP链接并下载文件
1. 项目缘起:从手动点击到自动化脚本的必然之路
在数据采集、文件归档或者日常运维工作中,我们经常会遇到一种看似简单却极其繁琐的场景:有一系列结构相似的网页,每个页面上都包含一个或多个指向FTP服务器文件的链接。我们的任务就是把这些文件全部下载下来。手动操作意味着你需要逐个打开网页,找到链接,复制,然后在下载工具或命令行中粘贴,循环往复。如果只有三五个页面还好,一旦面对几十上百个页面,这种重复劳动不仅效率低下,而且极易出错,比如漏掉某个页面,或者因为手滑复制错了链接。
这正是我最近接手的一个数据备份项目所面临的困境。客户提供了上百个产品资料页面,每个页面的源代码里都“隐藏”着一个指向FTP服务器的产品手册PDF链接。我的目标就是把这些PDF全部自动化地抓取下来。这个需求的核心可以拆解为三个动作: 爬取网页、解析链接、下载文件 。听起来像是爬虫的经典应用,但结合FTP协议,又有一些特别的细节需要注意。网上虽然有很多关于HTTP下载的教程,但专门针对“从网页中提取FTP链接并批量下载”的完整脚本方案却不多见,这也是我决定把这次实战经验整理成文的原因。
无论你是数据分析师需要定期抓取公开数据集,还是运维工程师要备份日志文件,亦或是像我需要处理批量资料,掌握编写这样一个脚本的能力,都能将你从重复的机械劳动中解放出来。接下来,我将详细拆解用Python实现这一过程的每一步,包括工具选型、核心逻辑、避坑指南以及完整的代码实现。
2. 技术栈选型与核心思路拆解
实现“从网页下载FTP文件”这个目标,我们需要一套组合工具。每种工具的选择背后都有其考量,并非随意拼凑。
2.1 为什么是Python?
首先,编程语言我选择了Python。这几乎是此类自动化任务的首选,原因有三点:第一,生态丰富,有大量成熟稳定的库支持网络请求、HTML解析和文件操作;第二,语法简洁,编写脚本速度快,即使非专业开发人员也相对容易上手和理解;第三,跨平台,脚本在Windows、Linux或macOS上都能运行,无需修改。
2.2 核心库的职责与选型理由
整个脚本的工作流像一条流水线,每个库负责一个环节:
-
requests/httpx(获取网页内容) :我们需要从互联网上把那些包含FTP链接的网页“拿”下来。requests库是Python中最著名的HTTP客户端库,其API设计极其人性化,几行代码就能完成复杂的网络请求。对于更现代或需要异步的场景,httpx也是一个优秀的选择。这里我们选择经典的requests,因为它足够简单稳定。注意 :有些网页可能需要处理Cookie、Session或简单的反爬机制(如User-Agent检查),
requests都能轻松应对。 -
BeautifulSoup4(解析HTML,提取链接) :拿到网页的HTML源代码后,它是一团混杂着标签、样式和脚本的文本。我们需要从中精准地找到那些FTP链接。BeautifulSoup4(简称bs4)就是一个强大的HTML/XML解析器。它允许我们使用像find_all(‘a‘)这样的方法来查找所有超链接标签,然后通过检查链接的href属性是否以ftp://开头,来筛选出我们需要的目标。 -
urllib/ftplib(处理FTP下载) :这是最关键的一步。提取出FTP链接后,我们需要连接FTP服务器并下载文件。Python标准库中提供了两种主要方式:-
urllib.request.urlretrieve:这是一个非常高级别的函数,你只需要给它一个FTP URL(如ftp://example.com/path/file.zip),它就能帮你完成连接、认证(如果支持匿名)、下载的全过程,并将文件保存到本地。对于简单的匿名FTP下载,这是最快捷的方式。 -
ftplib:这是一个更低级、更灵活的FTP客户端库。它允许你精确控制FTP连接的每一个环节:登录(支持用户名密码)、切换目录、设置传输模式(二进制或文本)、以及下载文件。当遇到需要认证、服务器有特殊目录结构或需要更稳健的错误处理时,ftplib是更好的选择。
-
如何选择? 如果所有FTP链接都是匿名可访问的,并且链接是完整的绝对路径, urllib 最简单。如果链接是相对路径,或者需要登录认证,那么就必须使用 ftplib 来手动构建完整的连接和下载逻辑。在本案例中,我遇到的链接都是完整的匿名FTP链接,因此为了演示的通用性,我会先展示 urllib 的简单方法,再深入讲解 ftplib 应对复杂场景的方案。
-
re(正则表达式,备用解析器) :虽然BeautifulSoup4是解析HTML的主力,但有时链接可能藏在JavaScript代码或某些特定的文本字符串中。正则表达式re模块作为一把“手术刀”,可以用来匹配和提取符合特定模式的文本(如所有包含ftp://的字符串),作为bs4的补充。
2.3 脚本的核心工作流程
整个脚本的逻辑链条非常清晰:
- 输入 :一个包含多个目标网页URL的列表。
- 循环处理每个网页 : a. 获取 :使用
requests.get()获取网页HTML。 b. 解析 :使用BeautifulSoup4解析HTML,找出所有<a>标签。 c. 过滤 :检查每个标签的href属性,筛选出以ftp://开头的链接。 d. 下载 :对于每个FTP链接,使用urllib或ftplib将其指向的文件下载到本地指定文件夹。 - 输出 :本地文件夹中下载好的所有文件,并在控制台输出下载日志(成功或失败)。
3. 环境准备与基础代码框架
在开始写核心逻辑之前,我们需要搭建好工作环境。这个过程本身也有一些小坑需要注意。
3.1 创建虚拟环境与安装依赖
我强烈建议为这个项目创建一个独立的Python虚拟环境。这可以避免与你系统上其他项目的库版本冲突。
# 在项目目录下,创建虚拟环境(这里以venv为例)
python -m venv ftp_downloader_env
# 激活虚拟环境
# 在Windows上:
ftp_downloader_env\Scripts\activate
# 在Linux/macOS上:
source ftp_downloader_env/bin/activate
# 激活后,命令行提示符前通常会显示环境名,如 (ftp_downloader_env)
接下来,安装我们所需的第三方库。 requests 和 beautifulsoup4 需要通过pip安装。
pip install requests beautifulsoup4
至于 urllib 和 re ,它们是Python的标准库,无需额外安装。 ftplib 也是标准库的一部分。
3.2 构建脚本的基础骨架
让我们先创建一个名为 download_ftp_from_pages.py 的Python文件,并搭建起最基本的程序结构。这个骨架包含了导入库、定义主函数和占位符。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
脚本:从一系列网页中自动下载FTP文件
作者:[你的名字]
功能:给定网页URL列表,自动爬取页面,提取所有FTP链接,并下载对应的文件到本地。
"""
import os
import sys
import time
import re
from urllib.parse import urljoin, urlparse
import urllib.request
import ftplib
from ftplib import FTP
import requests
from bs4 import BeautifulSoup
# 全局配置
DOWNLOAD_DIR = './downloaded_files' # 文件下载保存目录
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' # 模拟浏览器请求
REQUEST_TIMEOUT = 30 # 请求超时时间(秒)
RETRY_TIMES = 3 # 失败重试次数
def ensure_download_dir():
"""确保下载目录存在,如果不存在则创建"""
if not os.path.exists(DOWNLOAD_DIR):
os.makedirs(DOWNLOAD_DIR)
print(f"[信息] 创建下载目录: {DOWNLOAD_DIR}")
else:
print(f"[信息] 下载目录已存在: {DOWNLOAD_DIR}")
def download_file_via_urllib(ftp_url, local_path):
"""
使用 urllib 下载FTP文件(适用于匿名FTP)
:param ftp_url: 完整的FTP URL,例如 ftp://example.com/pub/file.zip
:param local_path: 本地保存路径
:return: 成功返回True,失败返回False
"""
# 具体实现见下一章节
pass
def download_file_via_ftplib(ftp_url, local_path, username='anonymous', password=''):
"""
使用 ftplib 下载FTP文件(支持认证)
:param ftp_url: 完整的FTP URL
:param local_path: 本地保存路径
:param username: FTP用户名,默认为匿名
:param password: FTP密码,默认为空
:return: 成功返回True,失败返回False
"""
# 具体实现见下一章节
pass
def extract_ftp_links_from_html(html_content, base_url):
"""
从HTML内容中提取所有FTP链接
:param html_content: 网页HTML文本
:param base_url: 网页的基准URL,用于处理相对路径(虽然FTP链接通常为绝对路径)
:return: FTP链接列表
"""
# 具体实现见下一章节
pass
def process_single_page(page_url):
"""
处理单个网页:抓取、解析、下载
:param page_url: 要处理的网页URL
:return: 该页面成功下载的文件列表
"""
# 具体实现见下一章节
pass
def main(page_urls):
"""
主函数
:param page_urls: 包含所有目标网页URL的列表
"""
print(f"[开始] 启动批量下载任务,共 {len(page_urls)} 个页面待处理。")
ensure_download_dir()
successful_downloads = []
failed_downloads = []
for idx, page_url in enumerate(page_urls, 1):
print(f"\n[处理中] ({idx}/{len(page_urls)}) 正在处理页面: {page_url}")
try:
files = process_single_page(page_url)
successful_downloads.extend(files)
print(f" -> 从本页面成功下载 {len(files)} 个文件。")
except Exception as e:
print(f" -> [错误] 处理页面失败: {e}")
failed_downloads.append(page_url)
# 输出总结报告
print(f"\n{'='*50}")
print("[任务完成] 总结报告:")
print(f" 成功处理的页面: {len(page_urls) - len(failed_downloads)}")
print(f" 成功下载的文件总数: {len(successful_downloads)}")
if failed_downloads:
print(f" 处理失败的页面: {len(failed_downloads)}")
for url in failed_downloads:
print(f" - {url}")
print(f"文件已保存至: {os.path.abspath(DOWNLOAD_DIR)}")
print('='*50)
if __name__ == '__main__':
# 示例:在这里替换成你的目标网页URL列表
target_pages = [
'http://example.com/products/page1.html',
'http://example.com/products/page2.html',
# ... 更多页面
]
if not target_pages:
print("[错误] 未提供任何目标页面URL。请在脚本的 `target_pages` 列表中添加。")
sys.exit(1)
main(target_pages)
这个框架已经具备了清晰的逻辑结构和良好的可扩展性。我们定义了配置常量、核心功能函数和主流程。接下来,我们将逐一填充那些 pass 占位符,实现真正的功能。
4. 核心功能实现:爬取、解析与下载
现在,我们来填充脚本的灵魂部分。我将分步实现提取链接和下载文件这两个核心功能,并分别给出 urllib 和 ftplib 两种下载方案的代码。
4.1 从HTML中精准提取FTP链接
extract_ftp_links_from_html 函数是信息提取的关键。它的目标是像筛子一样,从杂乱的HTML中捞出所有“金子”——也就是FTP链接。
def extract_ftp_links_from_html(html_content, base_url):
"""
从HTML内容中提取所有FTP链接
策略:先用BeautifulSoup查找所有<a>标签,再用正则表达式作为补充扫描全文。
"""
ftp_links = []
soup = BeautifulSoup(html_content, 'html.parser')
# 方法1:通过BeautifulSoup查找所有<a>标签的href属性
for a_tag in soup.find_all('a', href=True):
href = a_tag['href']
# 清洗和补全链接
full_url = urljoin(base_url, href)
# 判断是否为FTP链接
if full_url.lower().startswith('ftp://'):
if full_url not in ftp_links: # 去重
ftp_links.append(full_url)
# 方法2:使用正则表达式在整个HTML中搜索(应对链接不在<a>标签内的情况)
# 这个正则表达式匹配以 ftp:// 开头的URL,直到遇到空格、引号或HTML标签结束
ftp_url_pattern = re.compile(r'ftp://[^\s\'\"<>]+')
for match in ftp_url_pattern.findall(html_content):
full_url = urljoin(base_url, match)
if full_url not in ftp_links:
ftp_links.append(full_url)
print(f" [解析] 从HTML中提取到 {len(ftp_links)} 个FTP链接。")
return ftp_links
这里有几个重要的细节:
-
urljoin的作用 :网页中的链接可能是相对路径(如../files/manual.pdf)。urljoin函数能根据base_url(当前网页的URL)将这些相对路径补全为绝对路径。虽然FTP链接在网页中以绝对路径形式出现更常见,但这是一个好习惯,能让代码更健壮。 - 大小写处理 :使用
.lower().startswith('ftp://')来避免因FTP://或Ftp://等大小写不一致导致的遗漏。 - 去重 :同一个链接可能在页面中出现多次,去重可以避免重复下载。
- 正则表达式作为补充 :有些FTP链接可能被动态JavaScript生成,或者写在注释、JS变量里,
BeautifulSoup可能无法直接通过标签捕获。用正则表达式扫描全文文本是一种有效的补充手段,但要注意正则可能匹配到非链接的文本,精确度不如解析DOM高。
4.2 实现下载功能:两种方案对比
4.2.1 方案A:使用 urllib.request.urlretrieve (快速简单)
这是最简单粗暴的方法,适合匿名FTP且链接是完整URL的情况。
def download_file_via_urllib(ftp_url, local_path):
"""
使用 urllib 下载FTP文件
注意:此方法对于需要复杂认证或存在代理等网络环境时可能不适用。
"""
try:
# urlretrieve 会直接下载文件并保存到 local_path
# reporthook 可以用于显示下载进度(这里简单打印)
def progress_hook(count, block_size, total_size):
if total_size > 0:
percent = int(count * block_size * 100 / total_size)
sys.stdout.write(f"\r [下载] 进度: {percent}%")
sys.stdout.flush()
print(f" [下载] 开始下载: {ftp_url}")
urllib.request.urlretrieve(ftp_url, local_path, reporthook=progress_hook)
print(f"\n [成功] 文件已保存: {local_path}")
return True
except urllib.error.URLError as e:
print(f"\n [失败] URLError: {e.reason}")
except Exception as e:
print(f"\n [失败] 未知错误: {e}")
return False
优点 :代码极其简洁,无需处理FTP连接细节。 缺点 :
- 无法处理认证 :只能用于匿名FTP。
- 错误信息不直观 :
URLError可能包含多种原因,难以精准定位是网络问题、文件不存在还是权限问题。 - 灵活性差 :无法在下载前检查文件大小、修改传输模式等。
4.2.2 方案B:使用 ftplib (灵活强大)
这是更专业、更可控的方法。我们需要从FTP URL中解析出服务器地址、端口、路径,然后建立连接进行下载。
def download_file_via_ftplib(ftp_url, local_path, username='anonymous', password=''):
"""
使用 ftplib 下载FTP文件
"""
parsed_url = urlparse(ftp_url)
ftp_host = parsed_url.hostname
ftp_port = parsed_url.port or 21 # 默认FTP端口是21
remote_path = parsed_url.path # 服务器上的文件路径,如 /pub/data/file.zip
if not remote_path or remote_path == '/':
print(f" [失败] 无效的FTP文件路径: {ftp_url}")
return False
# 提取文件名
filename = os.path.basename(remote_path)
if not filename:
print(f" [失败] 无法从路径提取文件名: {remote_path}")
return False
ftp = None
try:
print(f" [连接] 正在连接 FTP 服务器 {ftp_host}:{ftp_port}...")
# 建立FTP连接
ftp = FTP()
ftp.connect(host=ftp_host, port=ftp_port, timeout=REQUEST_TIMEOUT)
ftp.login(user=username, passwd=password)
print(f" [登录] 成功登录 (用户: {username})")
# 切换到文件所在目录
remote_dir = os.path.dirname(remote_path)
if remote_dir:
ftp.cwd(remote_dir) # cwd: change working directory
print(f" [切换目录] 切换到远程目录: {remote_dir}")
# 获取文件大小(用于显示进度)
try:
file_size = ftp.size(filename)
if file_size:
print(f" [信息] 远程文件大小: {file_size} 字节")
except:
file_size = None # 某些服务器或文件可能不支持SIZE命令
# 以二进制模式下载文件
print(f" [下载] 开始下载远程文件: {filename}")
with open(local_path, 'wb') as local_file:
# 定义回调函数来显示进度
def callback(data):
local_file.write(data)
if file_size and hasattr(callback, 'downloaded_bytes'):
callback.downloaded_bytes += len(data)
percent = int(callback.downloaded_bytes * 100 / file_size)
sys.stdout.write(f"\r [下载] 进度: {percent}%")
sys.stdout.flush()
callback.downloaded_bytes = 0 # 初始化已下载字节数
# RETR: 下载文件
ftp.retrbinary(f'RETR {filename}', callback=callback, blocksize=8192)
print(f"\n [成功] 文件已保存至: {local_path}")
return True
except ftplib.error_perm as e:
# 权限错误,例如文件不存在、无访问权限
error_code = str(e).split()[0]
if error_code == '550':
print(f"\n [失败] 文件不存在或无法访问: {filename}")
else:
print(f"\n [失败] FTP权限错误: {e}")
except ftplib.all_errors as e:
print(f"\n [失败] FTP操作错误: {e}")
except Exception as e:
print(f"\n [失败] 未知错误: {e}")
finally:
# 确保关闭FTP连接
if ftp:
try:
ftp.quit()
except:
ftp.close()
return False
这段代码的详细解读与避坑点:
- 解析URL :
urlparse函数将ftp://ftp.example.com:21/pub/data/file.zip这样的URL拆解成hostname、port、path等组件。这是连接服务器的前提。 - 连接与登录 :
ftp.connect()建立TCP连接,ftp.login()进行身份验证。默认使用匿名登录,如果需要用户名密码,调用函数时传入即可。 - 切换目录 :
ftp.cwd()命令用于改变服务器上的当前工作目录。这是关键一步,因为RETR命令是基于当前目录的。我们从URL中解析出目录路径并切换过去。 - 获取文件大小 :
ftp.size()命令尝试获取文件大小,用于显示进度条。 注意 :这是一个非标准FTP命令,并非所有服务器都支持。如果服务器不支持,我们会捕获异常并继续,进度条将显示百分比但不显示具体字节数。 - 二进制模式下载 :
ftp.retrbinary()用于下载二进制文件(如图片、压缩包、PDF)。如果要下载文本文件,应使用ftp.retrlines(),但绝大多数情况下我们都用二进制模式,因为它能保证文件原样传输,不会因操作系统差异导致换行符等问题。 - 错误处理 :FTP错误主要分为
error_perm(永久性错误,如550 File not found)和error_temp(临时性错误)。我们特别处理了常见的550错误,给用户更清晰的提示。 - 资源清理 :在
finally块中确保关闭FTP连接,无论下载成功与否,避免连接泄漏。
4.3 整合:处理单个页面的完整流程
现在,我们把获取网页、提取链接和下载文件整合到 process_single_page 函数中。
def process_single_page(page_url):
"""
处理单个网页:抓取、解析、下载
"""
downloaded_files = []
headers = {'User-Agent': USER_AGENT}
# 步骤1: 获取网页内容(带重试机制)
html_content = None
for attempt in range(RETRY_TIMES):
try:
print(f" [请求] 获取页面内容 (尝试 {attempt + 1}/{RETRY_TIMES})...")
response = requests.get(page_url, headers=headers, timeout=REQUEST_TIMEOUT)
response.raise_for_status() # 如果状态码不是200,抛出HTTPError
html_content = response.text
print(f" [成功] 页面获取成功,大小: {len(html_content)} 字符")
break # 成功则跳出重试循环
except requests.exceptions.RequestException as e:
print(f" [失败] 请求出错: {e}")
if attempt == RETRY_TIMES - 1:
print(f" [放弃] 已达到最大重试次数,跳过此页面。")
return downloaded_files # 返回空列表
time.sleep(2) # 等待2秒后重试
# 步骤2: 提取FTP链接
ftp_urls = extract_ftp_links_from_html(html_content, page_url)
if not ftp_urls:
print(f" [信息] 本页面未发现FTP链接。")
return downloaded_files
# 步骤3: 遍历并下载每个FTP链接
for ftp_url in ftp_urls:
# 从FTP URL中提取出安全的文件名
parsed = urlparse(ftp_url)
remote_filename = os.path.basename(parsed.path)
if not remote_filename:
print(f" [跳过] 无法从URL获取文件名: {ftp_url}")
continue
# 清理文件名,移除可能引起问题的字符
safe_filename = re.sub(r'[<>:\"/\\|?*]', '_', remote_filename)
local_file_path = os.path.join(DOWNLOAD_DIR, safe_filename)
# 检查文件是否已存在,避免重复下载
if os.path.exists(local_file_path):
print(f" [跳过] 文件已存在: {safe_filename}")
downloaded_files.append(local_file_path)
continue
# 选择下载方法(这里以ftplib为例,因其更稳健)
success = download_file_via_ftplib(ftp_url, local_file_path)
if success:
downloaded_files.append(local_file_path)
return downloaded_files
这个函数加入了几个提升健壮性的关键设计:
- 重试机制 :网络请求可能因瞬时波动失败。我们设置了
RETRY_TIMES次重试,每次失败后等待片刻再试。 - 文件名安全处理 :从URL提取的文件名可能包含操作系统不允许的字符(如
:,?,*等)。我们使用正则表达式re.sub将这些字符替换为下划线。 - 去重检查 :下载前检查本地是否已存在同名文件,避免不必要的网络流量和时间消耗。
5. 实战进阶:处理复杂场景与性能优化
基础功能跑通后,我们会发现真实世界远比理想情况复杂。下面针对几个常见棘手场景,提供解决方案和优化思路。
5.1 场景一:FTP链接需要登录认证
我们的 download_file_via_ftplib 函数已经预留了 username 和 password 参数。但问题在于,如何将认证信息与不同的FTP链接对应起来?网页上的FTP链接本身通常不包含密码。
解决方案:建立映射关系 最实用的方法是在脚本中维护一个“服务器-凭证”的映射字典。在解析出FTP链接后,根据其主机名(hostname)来查找对应的用户名和密码。
# 在脚本开头配置部分添加
FTP_CREDENTIALS = {
'ftp.private-server.com': {'username': 'myuser', 'password': 'mypass123'},
'internal.data.com': {'username': 'backup', 'password': 'readonly2024'},
# 其他需要认证的服务器...
}
def get_credentials_for_host(hostname):
"""根据FTP主机名获取登录凭证"""
creds = FTP_CREDENTIALS.get(hostname)
if creds:
return creds['username'], creds['password']
# 默认返回匿名登录凭证
return 'anonymous', ''
然后在调用 download_file_via_ftplib 时,传入对应的凭证:
hostname = urlparse(ftp_url).hostname
username, password = get_credentials_for_host(hostname)
success = download_file_via_ftplib(ftp_url, local_file_path, username, password)
安全提示 :将密码明文写在代码中是极不安全的。在生产环境中,应该使用环境变量、加密的配置文件或密钥管理服务来存储敏感信息。
5.2 场景二:网页是动态加载的(AJAX/JavaScript)
requests 获取的是服务器的初始HTML响应。如果页面内容是由JavaScript在浏览器中动态渲染的(例如单页应用SPA),那么 requests 拿到的HTML里可能没有我们想要的FTP链接。
解决方案:使用Selenium模拟浏览器 Selenium 是一个自动化测试工具,它可以控制真实的浏览器(如Chrome、Firefox)来加载页面,等待JavaScript执行完毕,再获取完整的DOM内容。
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
def get_dynamic_page_content(url):
"""使用Selenium获取动态加载的页面内容"""
# 需要先下载对应浏览器的WebDriver(如chromedriver)
options = webdriver.ChromeOptions()
options.add_argument('--headless') # 无头模式,不显示浏览器窗口
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)
try:
driver.get(url)
# 显式等待,直到某个包含链接的特定元素加载出来
# 例如,等待一个ID为'content'的元素出现,最多等10秒
WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.ID, "content"))
)
# 获取页面源代码
html = driver.page_source
return html
finally:
driver.quit()
在 process_single_page 函数中,你可以根据URL特征判断是否使用Selenium。但请注意,Selenium速度远慢于 requests ,且需要额外安装和配置,应仅作为最后手段。
5.3 场景三:大规模下载与性能优化
当需要处理成千上万个文件时,串行下载会非常慢。我们可以从两个层面优化:
1. 多线程/多进程下载 Python的 concurrent.futures 模块可以方便地实现线程池,让多个文件同时下载。
from concurrent.futures import ThreadPoolExecutor, as_completed
def download_multiple_files_concurrently(ftp_url_list, max_workers=5):
"""并发下载多个FTP文件"""
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 提交下载任务到线程池
future_to_url = {executor.submit(download_single_file, url): url for url in ftp_url_list}
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
success = future.result()
# 处理结果...
except Exception as exc:
print(f'{url} 下载过程中产生异常: {exc}')
重要提醒 :FTP协议本身不是为高并发设计的。对同一个FTP服务器发起过多并发连接,可能会被服务器拒绝或封禁。 max_workers 不宜设置过大,通常3-5个线程足矣。更好的策略是针对不同的FTP服务器使用不同的连接池。
2. 断点续传 对于超大文件,网络中断可能导致前功尽弃。 ftplib 本身不直接支持断点续传,但我们可以通过组合命令实现。
原理是:先检查本地已下载文件的大小,然后使用 ftp.sendcmd('REST ' + local_size) 命令,告诉服务器从指定字节位置开始传输,最后使用 RETR 命令下载剩余部分。
def download_with_resume(ftp, remote_filename, local_path):
"""支持断点续传的下载"""
local_size = 0
mode = 'ab' # 追加模式,用于续传
if os.path.exists(local_path):
local_size = os.path.getsize(local_path)
print(f" 发现本地部分文件,大小: {local_size} 字节,尝试续传...")
with open(local_path, mode) as f:
# 定义回调函数
def callback(data):
f.write(data)
# 如果已有部分文件,发送REST命令
if local_size > 0:
ftp.sendcmd(f"REST {local_size}")
# 执行下载(从REST指定的位置开始)
ftp.retrbinary(f'RETR {remote_filename}', callback)
需要注意的是, 并非所有FTP服务器都支持 REST 命令 。在尝试续传前,最好先用 ftp.sendcmd('FEAT') 查看服务器支持的功能列表。
6. 完整脚本示例与使用指南
将上述所有模块组合起来,我们就得到了一个功能相对完整的脚本。以下是整合后的核心部分示意,以及如何使用它的指南。
6.1 脚本的配置与运行
-
编辑目标URL列表 :在脚本最底部的
__main__部分,找到target_pages列表,将你的目标网页URL填入。if __name__ == '__main__': target_pages = [ 'https://data.example.org/archive/2024/', 'https://software.mirror.list/downloads/', # 添加你的页面URL ] main(target_pages) -
配置下载目录和认证信息 :在脚本开头的全局配置部分,按需修改。
DOWNLOAD_DIR = './my_downloads' # 修改为你想要的目录 FTP_CREDENTIALS = {} # 如果需要,在这里添加服务器认证信息 -
运行脚本 :在命令行中,进入脚本所在目录,运行:
python download_ftp_from_pages.py
6.2 一个增强版的完整脚本框架
考虑到模块化和可维护性,一个更工程化的做法是将配置(如URL列表、凭证)放在单独的 config.py 或 config.yaml 文件中,并使用日志模块 logging 替代 print 来记录更结构化的运行信息。这里提供一个思路:
# config.yaml (示例)
target_pages:
- url: "https://site1.com/page1"
use_selenium: false
- url: "https://app.site2.com/dashboard"
use_selenium: true
ftp_credentials:
"ftp.private.com":
username: "user1"
password: "pass1"
download_dir: "./data"
max_workers: 3
在主脚本中,使用 yaml 或 json 库加载配置。同时,将主要的类或函数封装,使逻辑更清晰。
6.3 常见问题排查(Q&A)
在运行脚本时,你可能会遇到以下问题,这里给出排查思路:
-
Q: 脚本运行后没有任何文件下载,控制台显示“未发现FTP链接”。
- A1 : 检查目标网页是否真的包含
ftp://开头的链接。可以在浏览器中打开页面,查看源代码(Ctrl+U),搜索“ftp://”。 - A2 : 链接可能是动态加载的。尝试使用Selenium方案。
- A3 : 链接可能不是
<a>标签,而是其他形式(如<link>或纯文本)。尝试调整BeautifulSoup的查找条件,或加强正则表达式。
- A1 : 检查目标网页是否真的包含
-
Q: 下载文件时失败,提示“550 File not found”或“550 Permission denied”。
- A1 : “File not found”:检查FTP链接是否已失效,或者路径是否正确。有时网页上的链接可能是相对路径,我们的
urljoin逻辑可能没处理好。可以打印出准备下载的完整FTP URL进行核对。 - A2 : “Permission denied”:该文件需要特定权限。确认你是否使用了正确的用户名和密码登录。对于匿名FTP,有些目录或文件可能禁止匿名访问。
- A1 : “File not found”:检查FTP链接是否已失效,或者路径是否正确。有时网页上的链接可能是相对路径,我们的
-
Q: 下载速度非常慢。
- A1 : 可能是网络问题或FTP服务器限速。可以尝试在非高峰时段运行。
- A2 : 如果服务器在国外,网络延迟可能很高。考虑使用代理(需要在
requests和ftplib中配置,但注意代理对FTP的支持可能有限)。 - A3 : 检查是否错误地使用了文本模式(
retrlines)下载二进制文件,这会导致数据被错误转换,可能变慢或损坏。
-
Q: 下载到一半连接断开,如何继续?
- A : 实现完整的断点续传逻辑(如5.3节所述),并确保在异常捕获后保存已下载的部分。最简单的临时方案是:删除不完整的文件,重新运行脚本,依靠我们已有的“文件存在则跳过”逻辑来避免重复下载已成功的部分,但未完成的文件需要重新开始。
编写这样一个脚本,从零到一解决实际问题的过程,远比单纯调用一个现成工具更有价值。它迫使你深入理解HTTP、HTML、FTP等多种协议和技术的交互细节,并考虑网络异常、数据解析、资源管理等一系列工程问题。当你看到脚本自动将上百个文件井然有序地下载到本地时,那种效率提升带来的成就感,就是对这段编码时间最好的回报。希望这篇详尽的指南,能帮你顺利搭建起属于自己的自动化下载工具。
更多推荐


所有评论(0)