Python网络爬虫：高效抓取与管理网站文件实战指南61

找到文件链接后，下一步就是下载它们。对于大型文件，建议使用流式下载，以节省内存。import os
import shutil
import time
def download_file(file_url, save_directory="downloads", filename=None):
if not (save_directory):
(save_directory)
if filename is None:
# 从URL中提取文件名，或生成一个唯一的名称
filename = (urlparse(file_url).path)
if not filename: # 如果路径末尾没有文件名（例如 /download/），可以尝试从query string获取或生成
filename = f"downloaded_file_{int(())}{(file_url)[1]}" # 使用时间戳和扩展名
save_path = (save_directory, filename)
if (save_path):
print(f"文件 {filename} 已存在，跳过下载。")
return
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
try:
# 使用 stream=True 进行流式下载，适合大文件
with (file_url, headers=headers, stream=True, timeout=30) as r:
r.raise_for_status() # 检查HTTP响应状态
with open(save_path, 'wb') as f:
# 逐块写入文件
for chunk in r.iter_content(chunk_size=8192):
(chunk)
print(f"成功下载: {file_url} 到 {save_path}")
except as e:
print(f"下载 {file_url} 失败: {e}")
# 示例
# save_folder = "my_downloads"
# # file_links = ... (从上一步获取)
# # for link in file_links:
# # download_file(link, save_folder)

解释：

(save_directory)：如果下载目录不存在则创建。
(urlparse(file_url).path)：从URL中提取文件的实际名称，这是最常见和安全的方法。
stream=True和r.iter_content(chunk_size=8192)：requests库的流式下载机制，它不会一次性将整个文件载入内存，而是分块传输，大大降低了内存占用，特别适合下载大文件。
'wb'模式：以二进制写入模式打开文件。
文件去重：在下载前检查文件是否存在，避免重复下载。

四、实战案例：下载特定文件类型

现在，我们将以上步骤整合到一个完整的脚本中，演示如何从一个网页下载所有PDF文件。import requests
from bs4 import BeautifulSoup
from import urljoin, urlparse
import os
import shutil
import time
def get_html_content(url, headers):
try:
response = (url, headers=headers, timeout=10)
response.raise_for_status()
= response.apparent_encoding
return
except as e:
print(f"请求 {url} 失败: {e}")
return None
def find_file_links(html_content, base_url, file_extensions):
soup = BeautifulSoup(html_content, 'lxml')
file_urls = []
for a_tag in soup.find_all('a', href=True):
href = a_tag['href']
full_url = urljoin(base_url, href)

path = urlparse(full_url).path
if any(().endswith(ext) for ext in file_extensions):
(full_url)

return list(set(file_urls))
def download_file(file_url, save_directory, headers):
if not (save_directory):
(save_directory)
filename = (urlparse(file_url).path)
if not filename or ('?'): # 简单的文件名校验，避免URL末尾是查询参数
filename = f"downloaded_file_{int(())}{(file_url)[1].split('?')[0]}"
if not ('.')[-1]: # 如果仍然没有有效扩展名，默认给个txt
filename += ".txt"
save_path = (save_directory, filename)
if (save_path):
print(f"文件 {filename} 已存在，跳过下载。")
return
try:
with (file_url, headers=headers, stream=True, timeout=30) as r:
r.raise_for_status()
with open(save_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
(chunk)
print(f"成功下载: {file_url} 到 {save_path}")
except as e:
print(f"下载 {file_url} 失败: {e}")
def main():
target_url = "/documents/" # 替换为你要爬取的网页URL
output_folder = "downloaded_pdfs"
allowed_extensions = ['.pdf'] # 设定要下载的文件类型
default_headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
print(f"开始爬取页面: {target_url}")
html_content = get_html_content(target_url, default_headers)
if html_content:
print("解析页面，查找文件链接...")
found_links = find_file_links(html_content, target_url, allowed_extensions)

if found_links:
print(f"找到 {len(found_links)} 个 {', '.join(allowed_extensions)} 文件链接。")
for link in found_links:
download_file(link, output_folder, default_headers)
(1) # 增加延迟，避免对网站造成过大压力
else:
print(f"未找到任何 {', '.join(allowed_extensions)} 文件。")
else:
print("无法获取网页内容，程序终止。")
if __name__ == "__main__":
main()

请将`target_url`替换为您希望爬取的实际网页URL。

五、高级特性与最佳实践

为了使您的爬虫更加健壮、高效和负责任，可以考虑以下高级特性和最佳实践：

1. 遵守``协议

在爬取任何网站之前，务必检查其``文件（通常位于`/`）。这个文件指示了哪些路径允许爬取，哪些不允许。Python有专门的库可以解析它，例如`robotparser`模块（Python 3.8+）或第三方库如`robotexclusionrulesparser`。

2. 设置下载延时

频繁的请求可能会被网站视为攻击行为，导致IP被封禁。在每次请求之间添加随机或固定的延时（如`(1)`或`((1, 3))`）是基本的礼貌和自我保护。

3. 错误处理与日志记录

使用`try-except`块捕获网络错误、HTTP错误、解析错误等，并记录详细的日志。良好的日志可以帮助您调试问题和监控爬虫运行状态。

4. 处理动态加载内容（JavaScript）

如果目标网站的文件链接是通过JavaScript动态加载的，`requests`和`BeautifulSoup`将无法直接获取。此时，您需要使用像`Selenium`或`Playwright`这样的自动化浏览器工具来模拟用户行为，执行JavaScript，然后获取渲染后的页面内容。from selenium import webdriver
from import Service
from import Options
# ... (其他导入)
def get_dynamic_html_content(url):
chrome_options = Options()
chrome_options.add_argument("--headless") # 无头模式，不显示浏览器界面
chrome_options.add_argument("--disable-gpu")
chrome_options.add_argument("user-agent=Mozilla/5.0...") # 设置User-Agent
service = Service('/path/to/chromedriver') # 替换为你的chromedriver路径
driver = (service=service, options=chrome_options)

try:
(url)
(5) # 等待页面JS加载完成
return driver.page_source
except Exception as e:
print(f"使用Selenium请求 {url} 失败: {e}")
return None
finally:
()
# html_content = get_dynamic_html_content(target_url)

5. 异步爬取提升效率

对于需要同时下载大量文件的场景，可以使用Python的`asyncio`配合`aiohttp`库实现异步并发下载，大幅提高效率。这比简单的多线程/多进程更适合I/O密集型任务。import asyncio
import aiohttp
# ... (其他导入，BeautifulSoup, urlparse, os等)
async def async_download_file(session, file_url, save_directory, headers):
# ... (与上面的download_file类似，但使用await ())
pass
async def async_main():
target_url = "/documents/"
output_folder = "downloaded_pdfs"
allowed_extensions = ['.pdf']
default_headers = { ... }
html_content = get_html_content(target_url, default_headers) # 页面获取仍可以是同步
if html_content:
found_links = find_file_links(html_content, target_url, allowed_extensions)
if found_links:
async with (headers=default_headers) as session:
tasks = []
for link in found_links:
(async_download_file(session, link, output_folder, default_headers))
await (*tasks)
else:
print("未找到任何文件。")
else:
print("无法获取网页内容。")
# if __name__ == "__main__":
# (async_main())

6. 代理IP池

如果您的爬虫被目标网站封禁IP，可以考虑使用代理IP池。在`()`中通过`proxies`参数传递代理服务器信息。proxies = {
"http": "user:password@:8080",
"https": "user:password@:8080",
}
response = (url, headers=headers, proxies=proxies, timeout=10)

7. 文件命名与去重

确保下载的文件有合理的命名规则，避免文件重名覆盖。可以考虑在文件名中加入源URL的哈希值或日期时间戳，或者在文件存在时跳过下载。

六、法律与道德风险

再次强调，构建和运行网络爬虫涉及法律和道德风险：
版权问题： 未经授权下载受版权保护的内容可能触犯法律。
隐私侵犯： 爬取和存储个人敏感数据是严格禁止的。
服务条款： 许多网站的服务条款明确禁止自动化抓取。
服务器压力： 频繁或大规模的请求可能导致网站服务器过载，影响正常用户访问。

在任何情况下，请确保您的爬虫行为是合法、负责任且尊重的。

2025-10-18