如何使用Python遍历多页URL批量下载符合过滤规则的PDF文件

前置依赖安装 #

先执行命令安装所需第三方库：
pip install requests beautifulsoup4 tqdm

实现规则说明 #

• 分页遍历逻辑：基于给定的分页URL模板逐页请求，当返回页无有效搜索结果时自动终止遍历，无需手动配置总页数
• 链接检测逻辑：一级搜索结果页中直接指向PDF的链接直接进入下载校验流程；非PDF跳转链接自动发起二级请求，提取目标页内所有PDF资源链接
• 过滤规则：所有待下载PDF自动提取文件名，文件名以"INS"开头的文件直接跳过，不执行下载
• 容错逻辑：内置请求超时、重复文件跳过机制，避免因网络波动导致爬取中断

完整代码 #

import os
import requests
from urllib.parse import urljoin, unquote
from bs4 import BeautifulSoup
from tqdm import tqdm

# -------------------------- 配置项 可根据需求修改 --------------------------
START_YEAR = 2020  # 检索起始年份
END_YEAR = 2024    # 检索结束年份
SEARCH_KEYWORD = ""  # 搜索关键词，无关键词留空即可
SAVE_DIR = "./asn_pdfs"  # PDF本地保存路径
REQUEST_TIMEOUT = 15  # 单次请求超时时间 单位秒
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
    "Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7"
}
# ------------------------------------------------------------------------

# 初始化保存目录
os.makedirs(SAVE_DIR, exist_ok=True)
# 已下载文件集合 用于去重
downloaded_files = set(os.listdir(SAVE_DIR))
# 全局已爬取链接集合 避免重复请求二级页
crawled_pages = set()

def extract_pdf_from_page(page_url):
    """从指定页面提取所有符合规则的PDF下载链接"""
    pdf_links = []
    if page_url in crawled_pages:
        return pdf_links
    crawled_pages.add(page_url)
    
    try:
        resp = requests.get(page_url, headers=HEADERS, timeout=REQUEST_TIMEOUT)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, "html.parser")
        
        for a_tag in soup.find_all("a", href=True):
            href = a_tag["href"].strip()
            absolute_url = urljoin(page_url, href)
            # 识别PDF链接
            if absolute_url.lower().endswith(".pdf"):
                # 提取文件名 过滤INS开头的文件
                file_name = unquote(absolute_url.split("/")[-1])
                if file_name.startswith("INS"):
                    print(f"跳过规则过滤文件: {file_name}")
                    continue
                pdf_links.append((absolute_url, file_name))
    except Exception as e:
        print(f"页面解析失败 {page_url}: {str(e)}")
    
    return pdf_links

def download_pdf(pdf_url, file_name):
    """执行PDF下载"""
    if file_name in downloaded_files:
        return
    save_path = os.path.join(SAVE_DIR, file_name)
    
    try:
        resp = requests.get(pdf_url, headers=HEADERS, stream=True, timeout=REQUEST_TIMEOUT)
        resp.raise_for_status()
        total_size = int(resp.headers.get("content-length", 0))
        
        with open(save_path, "wb") as f, tqdm(
            desc=file_name,
            total=total_size,
            unit="iB",
            unit_scale=True,
            unit_divisor=1024,
            leave=False
        ) as pbar:
            for chunk in resp.iter_content(chunk_size=1024):
                if chunk:
                    size = f.write(chunk)
                    pbar.update(size)
        downloaded_files.add(file_name)
        print(f"下载完成: {file_name}")
    except Exception as e:
        print(f"下载失败 {pdf_url}: {str(e)}")
        if os.path.exists(save_path):
            os.remove(save_path)

if __name__ == "__main__":
    page_num = 1
    url_template = "https://www.asn.fr/recherche?filter_year[from]={}&filter_year[to]={}&limit=50&search_content_type=&search_text={}&sort_type=date&page={}"
    
    while True:
        current_page_url = url_template.format(START_YEAR, END_YEAR, SEARCH_KEYWORD, page_num)
        print(f"\n正在爬取第 {page_num} 页: {current_page_url}")
        
        try:
            resp = requests.get(current_page_url, headers=HEADERS, timeout=REQUEST_TIMEOUT)
            resp.raise_for_status()
            soup = BeautifulSoup(resp.text, "html.parser")
            result_links = soup.find_all("a", href=True)
            
            # 无有效结果时终止遍历
            if not result_links:
                print("已到达最后一页，爬取终止")
                break
            
            page_pdf_tasks = []
            for a_tag in result_links:
                href = a_tag["href"].strip()
                absolute_url = urljoin(current_page_url, href)
                # 跳过非http链接、锚点链接
                if not absolute_url.startswith("http") or "#" in absolute_url:
                    continue
                # 直接是PDF链接的直接加入任务
                if absolute_url.lower().endswith(".pdf"):
                    file_name = unquote(absolute_url.split("/")[-1])
                    if not file_name.startswith("INS"):
                        page_pdf_tasks.append((absolute_url, file_name))
                # 非PDF链接进入二级页提取PDF，仅爬取本站点链接避免跳转外部站
                elif "asn.fr" in absolute_url:
                    page_pdf_tasks.extend(extract_pdf_from_page(absolute_url))
            
            # 执行当前页所有下载任务
            for pdf_url, file_name in page_pdf_tasks:
                download_pdf(pdf_url, file_name)
            
            page_num += 1
        except Exception as e:
            print(f"第 {page_num} 页请求失败: {str(e)}，终止爬取")
            break

使用注意事项 #

• 首次运行前可根据自身需求修改代码开头配置块内的参数，默认检索2020-2024年全站无关键词的所有公开内容
• 脚本仅会爬取asn.fr域名下的页面，不会跳转至外部站点，避免无效请求
• 爬取过程中会自动跳过已下载文件，中途中断后重新运行会从断点位置继续，不会重复下载
• 若爬取速度较慢，可自行调整代码内的REQUEST_TIMEOUT参数，或添加代理配置

内容的提问来源于stack exchange，提问作者Ing DESIGN