Python requests批量抓取网站PDF并流式下载保存

脚本专家

一、需求与技术选型

PDF 版式固定、跨平台一致，适合严肃数据发布，但它是二进制资源，采集时必须在完整下载和内存占用之间做权衡。一个稳健的采集器要解决三件事：怎么找链接、怎么下载大文件、怎么在限流和网络抖动下保持稳定。技术选型上，requests + BeautifulSoup 学习成本低，适合百到千级的中小批量；httpx + parsel 支持异步；Scrapy 提供框架级并发和调度。本文以 requests + BeautifulSoup 为主线，其会话复用、流式传输、退避重试同样适用于 httpx 和 Scrapy。

依赖安装：

2. pip install requests beautifulsoup4 lxml
3. # 若需异步方案，可选用 httpx：
4. # pip install httpx

复制代码

建议使用虚拟环境隔离依赖，例如 python -m venv .venv && source .venv/bin/activate，并通过 requirements.txt 固化版本。

二、链接抽取与 URL 归一化

静态 HTML 中的 PDF 链接通常直接出现在 a href 中，用 BeautifulSoup 解析即可。若链接由 JavaScript 在客户端生成，例如滚动加载或接口分页，则静态解析拿不到，需要 Playwright/Selenium 执行渲染，或逆向其数据接口直接请求 JSON。本文聚焦静态 HTML。

网页里的 href 可能是相对路径、绝对路径或协议相对路径，直接拼接触发 404 或协议错误，必须用 urllib.parse.urljoin 归一化。链接还可能带查询参数，需剥离后作为文件名。

2. from urllib.parse import urljoin, urlparse
3. from pathlib import Path
5. def normalize_url(base: str, href: str) -> str:
6.     # 将任意形式的 href 解析为绝对 URL。
7.     return urljoin(base, href)
9. def safe_filename(pdf_url: str) -> str:
10.     # 从 URL 提取合法文件名：剥离查询串与片段，兜底默认名。
11.     name = urlparse(pdf_url).path.rstrip('/').split('/')[-1]
12.     name = name.split('?')[0].split('#')[0]
13.     return name or 'document.pdf'

复制代码

三、流式下载与类型校验

PDF 体积可能从几 KB 到数百 MB。若用 resp.content 一次性读入内存，大文件会打爆进程。正确做法是开启 stream=True，以固定分块 iter_content 写入磁盘，同时校验 Content-Type 或后缀，避免把 HTML 错误页当成 PDF 保存。

2. def is_pdf_response(url: str, content_type: str) -> bool:
3.     # 判断响应是否为 PDF：优先看 Content-Type，后缀兜底。
4.     return ('application/pdf' in content_type.lower()) or url.lower().endswith('.pdf')

复制代码

四、会话复用与传输层重试

反复调用 requests.get 会为每个请求新建 TCP 连接，带来握手开销。Session 能复用连接池，并在传输层挂载 urllib3.Retry，对 429/5xx 等可重试状态码自动指数退避，无需在业务代码里手写重试。非幂等请求重试可能产生副作用，因此这里只允许 GET 重试。

2. import logging
3. from pathlib import Path
4. from requests import Session
5. from requests.adapters import HTTPAdapter
6. from urllib3.util.retry import Retry
8. logger = logging.getLogger(__name__)
10. def build_session(proxies: dict | None = None, retries: int = 3, pool_maxsize: int = 10) -> Session:
11.     # 构造带连接池与传输层重试的会话。
12.     session = Session()
13.     if proxies:
14.         session.proxies.update(proxies)
15.     retry = Retry(
16.         total=retries,
17.         backoff_factor=0.5,
18.         status_forcelist=[429, 500, 502, 503, 504],
19.         allowed_methods=frozenset(['GET']),
20.     )
21.     adapter = HTTPAdapter(max_retries=retry, pool_connections=pool_maxsize, pool_maxsize=pool_maxsize)
22.     session.mount('http://', adapter)
23.     session.mount('https://', adapter)
24.     return session

复制代码

backoff_factor=0.5 表示第 n 次重试等待 0.5 * (2 ** n) 秒，覆盖 429、500、502、503、504 等典型限流与网关错误。

五、流式下载函数与并发模型

下载是 I/O 密集型任务，线程等待网络时让出 CPU，多线程能提升吞吐。这里使用 ThreadPoolExecutor 而非 asyncio，因为 requests 生态成熟、调试直观；若规模上探到万级，再迁移到 httpx + asyncio 收益更明显。

2. from requests import Session
4. def download_pdf(session: Session, url: str, dest: Path, *, chunk_size: int = 1 << 13,
5.                  timeout: tuple = (10, 60)) -> int:
6.     # 流式下载单个 PDF，返回写入字节数。
7.     headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
8.     with session.get(url, headers=headers, stream=True, timeout=timeout) as resp:
9.         resp.raise_for_status()
10.         content_type = resp.headers.get('Content-Type', '')
11.         if not is_pdf_response(url, content_type):
12.             raise ValueError(f'目标非 PDF 资源: Content-Type={content_type}')
13.         written = 0
14.         with open(dest, 'wb') as fp:
15.             for chunk in resp.iter_content(chunk_size=chunk_size):
16.                 if chunk:
17.                     fp.write(chunk)
18.                     written += len(chunk)
19.         logger.info('downloaded %s -> %s (%d bytes)', url, dest, written)
20.         return written

复制代码

2. from concurrent.futures import ThreadPoolExecutor, as_completed
4. def crawl_concurrent(session: Session, pdf_urls: list[str], save_dir: str,
5.                      *, max_workers: int = 8) -> list[str]:
6.     # 并发下载一批 PDF，返回成功落盘的路径列表。
7.     save_path = Path(save_dir)
8.     save_path.mkdir(parents=True, exist_ok=True)
9.     succeeded: list[str] = []
10.     with ThreadPoolExecutor(max_workers=max_workers) as pool:
11.         futures = {
12.             pool.submit(download_pdf, session, u, save_path / safe_filename(u)): u
13.             for u in pdf_urls
14.         }
15.         for fut in as_completed(futures):
16.             url = futures[fut]
17.             try:
18.                 fut.result()
19.                 succeeded.append(url)
20.             except Exception as exc:
21.                 logger.error('download failed: %s | %s', url, exc)
22.     logger.info('done: %d/%d succeeded', len(succeeded), len(pdf_urls))
23.     return succeeded

复制代码

chunk_size 默认为 8KB，timeout=(10, 60) 区分连接超时和读取超时，便于定位瓶颈。并发模型对比：串行 for 循环吞吐最低、复杂度最简单，适合少量文件调试；ThreadPoolExecutor 吞吐高、复杂度低，适合中小批量；httpx + asyncio 吞吐最高，适合中大批量高并发。

六、生产级加固：合规、限流与幂等

任何采集行为都应以 robots.txt 为合规基线。urllib.robotparser 可解析并判断路径是否允许抓取，解析失败时保守放行。

2. from urllib.robotparser import RobotFileParser
3. from urllib.parse import urljoin
5. def is_allowed(url: str, user_agent: str = '*') -> bool:
6.     # 依据 robots.txt 判断该 URL 是否允许抓取；解析失败则保守放行。
7.     rp = RobotFileParser()
8.     rp.set_url(urljoin(url, '/robots.txt'))
9.     try:
10.         rp.read()
11.     except Exception:
12.         return True
13.     return rp.can_fetch(user_agent, url)

复制代码

即便目标允许抓取，过高频率仍可能构成 DoS 风险。工程上可在请求前施加带抖动的最小间隔，配合 Retry 的退避重试。

2. import time, random
4. def throttle(min_interval: float = 0.2) -> None:
5.     # 在请求前调用，施加带抖动的最小间隔。
6.     time.sleep(min_interval + random.uniform(0, min_interval))

复制代码

单 IP 采集规模扩大后容易触发频率阈值被封禁。代理 IP 池把请求分散到多个出口 IP，降低单 IP 命中率。原文推荐亿牛云代理，提供 HTTP/HTTPS 隧道，支持用户名密码鉴权与按请求自动换 IP。接入只需在会话层配置 proxies：

2. proxies = {
3.     'http': 'http://用户名:密码@proxy.16yun.cn:端口',
4.     'https': 'http://用户名:密码@proxy.16yun.cn:端口',
5. }
6. session = build_session(proxies=proxies)

复制代码

代理、随机 User-Agent、速率抖动三者叠加，可应对多数公开站点的反爬策略。亿牛云隧道模式的自动换 IP 能让出口 IP 在请求间自然轮换，配合 Retry 退避降低封禁概率。

生产环境难免中断，理想采集器应可重跑且结果一致。下载前校验目标文件是否已存在且完整，存在则跳过；也可用 ETag 或 Content-Length 做一致性比对，或在支持 Range 的源站实现断点续传。

2. def should_skip(dest: Path, expected_size: int | None = None) -> bool:
3.     # 幂等判断：文件已存在且大小匹配则跳过。
4.     if not dest.exists():
5.         return False
6.     if expected_size is not None and dest.stat().st_size != expected_size:
7.         return False
8.     return True

复制代码

把 should_skip 放在 download_pdf 之前调用，脚本就能在中断后只补缺失项，抓取成千上万个文件也更稳。

七、总结

整条采集管道的分水岭在四处：传输层用 Session 复用连接池、用 Retry 处理限流与抖动；内存模型用 stream=True + iter_content 流式落盘，大文件下内存恒定；健壮性用 Content-Type 校验防错存、ThreadPoolExecutor 并发提速、异常隔离保证单点失败不拖垮全局；可持续性用 robots.txt 合规基线、速率抖动、亿牛云代理分散出口、幂等续传，让采集器长期稳定合法运行。若目标站点是 JavaScript 动态渲染，下一步可引入 Playwright 执行无头浏览器，或逆向其数据接口直连 JSON。