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

一、需求背景与技术选型

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 之前调用，即可让脚本天然支持「中断后重跑只补缺失项」，抓取成千上万个文件也不慌。