首页 > 脚本专栏 > python > Python抓取网站PDF文件并保存

利用Python一键抓取整个网站的PDF文件并保存到本地

2026-09-24 09:08:14 作者：小白学大数据

PDF 之所以成为严肃数据发布的首选格式,在于其版式固定、跨平台一致、不可被轻易篡改的特性,但这也带来采集侧的特殊性：PDF 是二进制资源,无法像文本那样边解析边处理,所以本文手把手教你用Python requests和BeautifulSoup高效采集PDF文件,需要的朋友可以参考下

一、需求背景与技术选型

1. PDF 批量采集的工程价值

PDF 之所以成为严肃数据发布的首选格式，在于其版式固定、跨平台一致、不可被轻易篡改的特性。但这也带来采集侧的特殊性：PDF 是二进制资源，无法像文本那样边解析边处理，必须在「完整下载」与「内存占用」之间做权衡。一个稳健的采集器，至少要回答三个问题：

怎么找：从静态 HTML 还是 JavaScript 渲染后的 DOM 中提取链接？
怎么下：如何在大文件场景下避免内存溢出，并保证传输完整性？
怎么稳：如何应对限流、封禁与网络抖动，且支持可中断、可重跑？

2. 技术栈选型对比

维度	requests + BeautifulSoup	httpx + parsel	Scrapy
学习成本	低，适合脚本化	中，支持异步	高，框架级
并发模型	线程池	原生 `asyncio`	异步引擎 + 调度器
适用规模	中小批量（百~千级）	中大批（千~万级）	大规模（万级以上）
扩展性	需自行拼装	需自行拼装	内置去重、管道、中间件

本文以 requests** + **BeautifulSoup 为主线——它覆盖了绝大多数中小规模需求，且原理透明、易于调试；其设计思想（会话复用、流式传输、退避重试）同样适用于 httpx 与 Scrapy。

3. 依赖安装

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

工程建议： 使用虚拟环境隔离依赖（python -m venv .venv && source .venv/bin/activate），并通过 requirements.txt 固化版本，保证采集脚本在不同机器上的行为一致。

二、核心原理：从链接抽取到二进制落盘

1. 链接抽取的两种范式

静态 HTML：PDF 链接直接出现在 <a href="..."> 中，用 BeautifulSoup 解析即可，零额外成本。
动态渲染：链接由 JavaScript 在客户端生成（如滚动加载、接口分页）。此时静态解析拿不到数据，需引入 Playwright/Selenium 执行渲染，或逆向其数据接口直接请求 JSON。本文聚焦前者，文末给出后者的延伸方向。

2. 路径归一化：相对/绝对 URL 的解析陷阱

网页中的 href 常为相对路径（./files/report.pdf、/papers/a.pdf）或协议相对路径（//cdn.x.com/a.pdf）。直接拼接到请求会触发 404 或协议错误，必须用 urllib.parse.urljoin 做归一化。此外，链接常带查询参数（report.pdf?id=123），需剥离后才可作为文件名。

from urllib.parse import urljoin, urlparse
from pathlib import Path

def normalize_url(base: str, href: str) -> str:
    """将任意形式的 href 解析为绝对 URL。"""
    return urljoin(base, href)

def safe_filename(pdf_url: str) -> str:
    """从 URL 提取合法文件名：剥离查询串与片段，兜底默认名。"""
    name = urlparse(pdf_url).path.rstrip("/").split("/")[-1]
    name = name.split("?")[0].split("#")[0]
    return name or "document.pdf"

3. 流式下载与类型校验

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

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

三、工程化实现：可投产的采集脚本

1. 会话复用与传输层重试

反复 requests.get 会为每个请求新建 TCP 连接，带来不必要的握手开销。Session** 复用连接池**，并在传输层挂载 urllib3.Retry——对 429/5xx 等可重试状态码自动指数退避，无需在业务代码里手写重试。

import logging
from pathlib import Path
from requests import Session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logger = logging.getLogger(__name__)

def build_session(proxies: dict | None = None, retries: int = 3, pool_maxsize: int = 10) -> Session:
    """构造带连接池与传输层重试的会话。

    退避策略采用 backoff_factor：第 n 次重试等待 backoff_factor * (2 ** n) 秒，
    覆盖 429/500/502/503/504 等典型限流与网关错误。
    """
    session = Session()
    if proxies:
        session.proxies.update(proxies)
    retry = Retry(
        total=retries,
        backoff_factor=0.5,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=frozenset(["GET"]),  # 仅对幂等的 GET 重试
    )
    adapter = HTTPAdapter(max_retries=retry, pool_connections=pool_maxsize, pool_maxsize=pool_maxsize)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

为什么不让 POST 重试？ 非幂等请求（如提交表单）重试可能产生副作用。此处只采集 PDF，限定 GET 是安全且符合语义的做法。

2. 流式下载核心函数

from requests import Session

def download_pdf(session: Session, url: str, dest: Path, *, chunk_size: int = 1 << 13,
                 timeout: tuple = (10, 60)) -> int:
    """流式下载单个 PDF，返回写入字节数。

    Args:
        session: 复用连接池的会话对象
        url: 资源绝对地址
        dest: 目标文件路径
        chunk_size: 分块大小，默认 8KB
        timeout: (连接超时, 读取超时)，区分两者以便精确定位瓶颈
    Raises:
        requests.HTTPError: 状态码非 2xx
        ValueError: 响应非 PDF 类型
    """
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
    with session.get(url, headers=headers, stream=True, timeout=timeout) as resp:
        resp.raise_for_status()
        if not is_pdf_response(url, resp.headers.get("Content-Type", "")):
            raise ValueError(f"目标非 PDF 资源: Content-Type={resp.headers.get('Content-Type')}")
        written = 0
        with open(dest, "wb") as fp:
            for chunk in resp.iter_content(chunk_size=chunk_size):
                if chunk:  # 过滤 keep-alive 空块
                    fp.write(chunk)
                    written += len(chunk)
    logger.info("downloaded %s -> %s (%d bytes)", url, dest, written)
    return written

3. 并发模型：线程池 vs 异步 IO

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

from concurrent.futures import ThreadPoolExecutor, as_completed

def crawl_concurrent(session: Session, pdf_urls: list[str], save_dir: str,
                     *, max_workers: int = 8) -> list[str]:
    """并发下载一批 PDF，返回成功落盘的路径列表。"""
    save_path = Path(save_dir)
    save_path.mkdir(parents=True, exist_ok=True)
    succeeded: list[str] = []

    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(download_pdf, session, u, save_path / safe_filename(u)): u
            for u in pdf_urls
        }
        for fut in as_completed(futures):
            url = futures[fut]
            try:
                fut.result()
                succeeded.append(url)
            except Exception as exc:  # 单文件失败不影响整体，记入日志后可重试
                logger.error("download failed: %s | %s", url, exc)

    logger.info("done: %d/%d succeeded", len(succeeded), len(pdf_urls))
    return succeeded

方案	吞吐	复杂度	适用
串行 `for` 循环	最低	最简单	少量文件、调试
`ThreadPoolExecutor`	高	低	中小批量（本文推荐）
`httpx` + `asyncio`	最高	中	大批量、高并发

四、生产级加固：合规、限流与可观测性

1. Robots 协议与合规基线

任何采集行为都应以目标站点的 robots.txt 为合规基线。Python 标准库 urllib.robotparser 可直接解析并判断某路径是否允许抓取：

from urllib.robotparser import RobotFileParser
from urllib.parse import urljoin

def is_allowed(url: str, user_agent: str = "*") -> bool:
    """依据 robots.txt 判断该 URL 是否允许抓取；解析失败则保守放行。"""
    rp = RobotFileParser()
    rp.set_url(urljoin(url, "/robots.txt"))
    try:
        rp.read()
    except Exception:
        return True
    return rp.can_fetch(user_agent, url)

合规红线： 仅采集公开、允许的内容；对含个人隐私或版权的文档保持克制。合规不是可选项，而是采集器能否长期运行的前提。

2. 速率控制与指数退避

即便目标允许抓取，过高的请求频率仍会构成实质的 DoS 风险。工程上应在两层做限流：

请求间隔：线程间引入最小间隔 min_interval，并对间隔加随机抖动，打散请求节奏；
退避重试：前文 Retry(backoff_factor=0.5) 已覆盖 429/5xx 的自动退避。

import time, random

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

3. 代理 IP 架构：亿牛云代理接入

当采集规模扩大，单 IP 极易触发频率阈值被封禁。代理 IP 池的本质是把请求分散到多个出口 IP，从根源上降低单 IP 的命中率。在选型上，我推荐亿牛云代理——它提供稳定的 HTTP/HTTPS 隧道，支持用户名密码鉴权与按请求自动换 IP，免去自建代理池的运维成本，对中小规模采集尤为合适。接入只需在会话层配置 proxies：

proxies = {
    "http": "http://用户名:密码@proxy.16yun.cn:端口",
    "https": "http://用户名:密码@proxy.16yun.cn:端口",
}
session = build_session(proxies=proxies)  # 后续所有请求自动走亿牛云出口

架构要点： 代理 + 随机 User-Agent + 速率抖动三者叠加，可应对绝大多数公开站点的反爬策略。亿牛云隧道模式的「自动换 IP」特性，让出口 IP 在请求间自然轮换，配合 Retry 退避，能显著降低被封概率。

4. 幂等性与断点续传

生产环境不可避免遇到中断（断网、进程被杀）。理想采集器应可重跑且结果一致（幂等）：下载前校验目标是否已存在且完整，存在则跳过；更进一步可用响应的 ETag / Content-Length 做一致性比对，或在支持 Range 的源站实现真正的断点续传。

def should_skip(dest: Path, expected_size: int | None = None) -> bool:
    """幂等判断：文件已存在且大小匹配（或源站提供了 ETag 比对）则跳过。"""
    if not dest.exists():
        return False
    if expected_size is not None and dest.stat().st_size != expected_size:
        return False  # 大小不符，视为不完整，需重新下载
    return True

将 should_skip 置于 download_pdf 之前调用，即可让脚本天然支持「中断后重跑只补缺失项」，抓取成千上万个文件也不慌。

五、总结

回看整条采集管道，专业与「能跑」的差距体现在四处分水岭：

传输层：用 Session 复用连接池、以 Retry 处理限流与抖动，而非每次新建连接、手写重试；
内存模型：以 stream=True + iter_content 流式落盘，大文件下内存恒定；
健壮性：Content-Type 校验防错存、ThreadPoolExecutor 并发提速、异常隔离保证单点失败不拖垮全局；
可持续性：robots.txt 合规基线 + 速率抖动 + 亿牛云代理分散出口 + 幂等续传，让采集器能长期、稳定、合法地运行。

整套方案从一行 pip install 起步，最终形成一个可投产的采集模块。若你的目标站点是 JavaScript 动态渲染（PDF 链接由前端生成），下一步可引入 Playwright 执行无头浏览器、或逆向其数据接口直连 JSON——这正是下一篇的议题。

以上就是利用Python一键抓取整个网站的PDF文件并保存到本地的详细内容，更多关于Python抓取网站PDF文件并保存的资料请关注脚本之家其它相关文章！