一、引言在信息爆炸的时代互联网上存在着海量的PDF文档资源——学术论文、行业报告、技术手册、电子书籍等。无论是研究人员需要批量获取文献资料还是数据分析师需要收集行业报告亦或是普通用户希望整理自己的电子图书馆能够高效、稳定地从网络资源站批量下载PDF文件都是一项非常实用的技能。本文将带领读者从零开始使用Python开发一个功能完备的PDF文档批量下载器。我们将深入探讨requests库的核心用法、文件流式写入的优化策略、进度条显示的多种实现方式以及重定向跟踪和大文件断点续传这两个高级话题。通过本文的学习您不仅能够掌握编写专业级爬虫下载工具的技能更能深入理解HTTP协议在文件传输中的实际应用。目录一、引言二、开发环境与准备工作2.1 Python环境配置2.2 必需的第三方库2.3 项目结构规划三、基础下载功能实现3.1 最简化的PDF下载器3.2 添加请求头和重定向处理3.3 从URL中智能提取文件名四、流式写入与进度条显示4.1 流式下载原理4.2 使用tqdm实现专业进度条4.3 自定义进度回调函数五、断点续传机制5.1 HTTP Range协议5.2 实现断点续传5.3 校验文件完整性六、链接解析与批量下载6.1 从网页中提取PDF链接6.2 条件过滤与筛选6.3 批量下载管理器七、高级功能与优化7.1 反爬虫策略应对7.2 代理IP支持7.3 数据库存储与任务管理7.4 日志系统八、完整实战案例8.1 案例从学术论文网站批量下载PDF8.2 配置文件管理九、性能优化与最佳实践9.1 并发下载的性能考量9.2 内存优化9.3 错误处理与重试策略9.4 测试用例十、常见问题与解决方案10.1 SSL证书验证问题10.2 中文文件名乱码10.3 大文件内存不足10.4 服务器限制文件访问十一、总结与展望未来改进方向二、开发环境与准备工作2.1 Python环境配置首先我们需要确保Python环境已经正确安装。推荐使用Python 3.8及以上版本以获得更好的异步支持和类型提示功能。可以通过以下命令检查Python版本bashpython --version2.2 必需的第三方库本项目需要以下核心库的支持python# requirements.txt requests2.28.0 tqdm4.64.0 fake-useragent1.4.0 retry0.9.2requests最流行的HTTP客户端库用于发送网络请求tqdm强大的进度条显示工具让下载过程直观可见fake-useragent随机生成User-Agent降低被反爬的风险retry提供重试机制增强程序的健壮性安装命令bashpip install -r requirements.txt2.3 项目结构规划一个良好的项目结构有助于代码的维护和扩展textpdf_downloader/ ├── src/ │ ├── __init__.py │ ├── downloader.py # 核心下载逻辑 │ ├── parser.py # 链接解析模块 │ ├── utils.py # 工具函数 │ └── config.py # 配置管理 ├── downloads/ # 下载文件存储目录 ├── logs/ # 日志文件目录 ├── tests/ # 单元测试 ├── main.py # 程序入口 └── requirements.txt三、基础下载功能实现3.1 最简化的PDF下载器让我们从最基础的版本开始逐步构建完整功能pythonimport requests import os def simple_download(url, save_path): 最简单的PDF下载函数 try: response requests.get(url, streamTrue) response.raise_for_status() # 检查请求是否成功 with open(save_path, wb) as f: for chunk in response.iter_content(chunk_size8192): if chunk: f.write(chunk) print(f下载完成{save_path}) return True except Exception as e: print(f下载失败{e}) return False这个简单版本虽然能工作但存在几个明显的不足没有处理重定向没有进度显示不支持断点续传没有异常重试机制3.2 添加请求头和重定向处理在实际应用中很多PDF链接会经过重定向才能到达真实的文件地址。requests库默认会自动处理重定向但我们需要获取最终的URL以确保文件名正确pythondef get_final_url(url, max_redirects5): 获取重定向后的最终URL session requests.Session() session.max_redirects max_redirects try: response session.get(url, streamTrue, allow_redirectsTrue) return response.url, response except requests.exceptions.TooManyRedirects: print(f重定向次数过多{url}) return url, None3.3 从URL中智能提取文件名PDF文件的命名往往隐藏在URL中或Content-Disposition头部pythonimport re from urllib.parse import unquote def extract_filename(response, url): 从响应或URL中提取文件名 # 尝试从Content-Disposition获取 content_disposition response.headers.get(Content-Disposition) if content_disposition: filename_match re.search(rfilename[^;]*[\\]?([^\\;]), content_disposition) if filename_match: filename unquote(filename_match.group(1)) return filename if filename.lower().endswith(.pdf) else filename .pdf # 从URL路径中提取 parsed_url url.split(?)[0] # 去除查询参数 filename parsed_url.split(/)[-1] if filename: filename unquote(filename) if not filename.lower().endswith(.pdf): filename .pdf return filename # 默认名称 return unknown.pdf四、流式写入与进度条显示4.1 流式下载原理当我们下载大文件时如果一次性将整个文件加载到内存中不仅会消耗大量内存还可能导致程序崩溃。流式下载Streaming Download通过分块读取和写入的方式有效解决了这个问题pythondef stream_download(url, save_path, chunk_size8192): 流式下载大文件 response requests.get(url, streamTrue) total_size int(response.headers.get(content-length, 0)) with open(save_path, wb) as f: downloaded 0 for chunk in response.iter_content(chunk_sizechunk_size): if chunk: f.write(chunk) downloaded len(chunk) # 计算下载进度 if total_size 0: progress (downloaded / total_size) * 100 print(f\r下载进度{progress:.2f}%, end) print() # 换行4.2 使用tqdm实现专业进度条tqdm库提供了更加美观和功能丰富的进度条显示pythonfrom tqdm import tqdm def download_with_progress(url, save_path): 使用tqdm显示下载进度 response requests.get(url, streamTrue) total_size int(response.headers.get(content-length, 0)) # 使用tqdm创建进度条 with open(save_path, wb) as f: with tqdm(totaltotal_size, unitB, unit_scaleTrue, descos.path.basename(save_path)) as pbar: for chunk in response.iter_content(chunk_size8192): if chunk: f.write(chunk) pbar.update(len(chunk))4.3 自定义进度回调函数为了更好地控制进度显示我们可以实现自定义的回调函数pythonclass ProgressCallback: def __init__(self, total_size, descDownloading): self.total_size total_size self.downloaded 0 self.desc desc self.start_time time.time() def update(self, chunk_size): self.downloaded chunk_size if self.total_size 0: percent (self.downloaded / self.total_size) * 100 speed self.downloaded / (time.time() - self.start_time) remaining (self.total_size - self.downloaded) / speed if speed 0 else 0 print(f\r{self.desc}: {percent:.1f}% | f{self.downloaded/1024/1024:.1f}MB/{self.total_size/1024/1024:.1f}MB | f速度: {speed/1024:.1f}KB/s | f剩余: {remaining:.1f}s, end)五、断点续传机制5.1 HTTP Range协议断点续传的核心是HTTP协议的Range头部它允许客户端请求资源的特定范围textRange: bytes0-999 # 请求前1000个字节 Range: bytes1000-1999 # 请求第1000到1999个字节 Range: bytes5000- # 请求从5000字节到末尾 Range: bytes-500 # 请求最后500个字节服务器响应时会返回状态码206 Partial Content并在Content-Range头部中说明返回的数据范围。5.2 实现断点续传完整实现断点续传需要处理多个边界情况pythonimport os import requests from tqdm import tqdm def resume_download(url, save_path, max_retries3): 支持断点续传的下载函数 # 检查已下载部分 temp_path save_path .temp downloaded_size 0 if os.path.exists(temp_path): downloaded_size os.path.getsize(temp_path) print(f检测到已下载 {downloaded_size/1024/1024:.2f} MB继续下载...) # 发送带有Range的请求 headers {} if downloaded_size 0: headers[Range] fbytes{downloaded_size}- for attempt in range(max_retries): try: response requests.get(url, headersheaders, streamTrue) # 检查服务器是否支持断点续传 if downloaded_size 0 and response.status_code 416: # Range请求超出范围文件可能已下载完成 print(文件似乎已经完整下载) if os.path.exists(temp_path): os.rename(temp_path, save_path) return True if response.status_code not in [200, 206]: print(f请求失败状态码{response.status_code}) continue total_size downloaded_size int(response.headers.get(content-length, 0)) # 使用ab模式追加写入 mode ab if downloaded_size 0 else wb with open(temp_path, mode) as f: with tqdm(totaltotal_size, initialdownloaded_size, unitB, unit_scaleTrue, descos.path.basename(save_path)) as pbar: for chunk in response.iter_content(chunk_size8192): if chunk: f.write(chunk) pbar.update(len(chunk)) # 下载完成重命名文件 os.rename(temp_path, save_path) return True except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: print(f连接错误 (尝试 {attempt1}/{max_retries}): {e}) if attempt max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避 return False5.3 校验文件完整性断点续传后我们需要验证下载的PDF文件是否完整pythonimport hashlib def verify_file(file_path, expected_hashNone, hash_typemd5): 验证文件完整性 if not os.path.exists(file_path): return False if expected_hash: hash_func hashlib.new(hash_type) with open(file_path, rb) as f: for chunk in iter(lambda: f.read(8192), b): hash_func.update(chunk) actual_hash hash_func.hexdigest() return actual_hash.lower() expected_hash.lower() # 简单验证检查PDF文件头 with open(file_path, rb) as f: header f.read(5) return header b%PDF-六、链接解析与批量下载6.1 从网页中提取PDF链接在实际应用中我们通常需要从网页中提取所有PDF链接pythonfrom bs4 import BeautifulSoup import re def extract_pdf_links(html_content, base_url): 从HTML内容中提取PDF链接 soup BeautifulSoup(html_content, html.parser) pdf_links [] # 查找所有a标签 for a_tag in soup.find_all(a, hrefTrue): href a_tag[href] # 检查链接是否指向PDF文件 if href.lower().endswith(.pdf) or .pdf? in href.lower(): # 处理相对路径 full_url requests.compat.urljoin(base_url, href) pdf_links.append({ url: full_url, text: a_tag.get_text(stripTrue) or os.path.basename(full_url) }) # 也检查一些可能是PDF链接的文本模式 text soup.get_text() pdf_urls re.findall(rhttps?://[^\s]\.pdf, text) for url in pdf_urls: if not any(link[url] url for link in pdf_links): pdf_links.append({url: url, text: os.path.basename(url)}) return pdf_links6.2 条件过滤与筛选很多时候我们需要按照特定条件筛选PDF文件pythondef filter_pdf_links(links, filters): 根据条件筛选PDF链接 filters: { keywords: [report, annual], # 包含关键词 exclude: [sample, draft], # 排除关键词 min_size: 1024*1024, # 最小文件大小1MB max_size: 100*1024*1024, # 最大文件大小100MB date_after: 2023-01-01, # 发布日期之后 } filtered [] for link in links: # 关键词过滤 if filters.get(keywords): if not any(kw.lower() in link[url].lower() or kw.lower() in link[text].lower() for kw in filters[keywords]): continue # 排除关键词 if filters.get(exclude): if any(kw.lower() in link[url].lower() or kw.lower() in link[text].lower() for kw in filters[exclude]): continue filtered.append(link) return filtered6.3 批量下载管理器设计一个批量下载管理器支持并发下载和队列管理pythonfrom concurrent.futures import ThreadPoolExecutor, as_completed import threading class BatchDownloader: def __init__(self, max_workers3, save_dirdownloads): self.max_workers max_workers self.save_dir save_dir self.success_count 0 self.fail_count 0 self.lock threading.Lock() # 创建保存目录 os.makedirs(save_dir, exist_okTrue) def download_task(self, url, filenameNone): 单个下载任务 try: if not filename: filename extract_filename(url) save_path os.path.join(self.save_dir, filename) # 如果文件已存在跳过 if os.path.exists(save_path) and os.path.getsize(save_path) 0: print(f文件已存在跳过{filename}) with self.lock: self.success_count 1 return True success resume_download(url, save_path) with self.lock: if success: self.success_count 1 else: self.fail_count 1 return success except Exception as e: print(f下载失败 {filename}: {e}) with self.lock: self.fail_count 1 return False def batch_download(self, urls, filenamesNone): 批量下载 tasks [] if filenames and len(filenames) len(urls): tasks list(zip(urls, filenames)) else: tasks [(url, None) for url in urls] with ThreadPoolExecutor(max_workersself.max_workers) as executor: futures [ executor.submit(self.download_task, url, name) for url, name in tasks ] for future in as_completed(futures): future.result() # 等待任务完成捕获异常 print(f\n下载完成成功{self.success_count}失败{self.fail_count})七、高级功能与优化7.1 反爬虫策略应对在实际爬取过程中我们可能会遇到各种反爬虫机制pythonfrom fake_useragent import UserAgent import time import random class AntiScrapingManager: def __init__(self): self.ua UserAgent() self.session requests.Session() def get_headers(self, refererNone): 生成随机请求头 headers { User-Agent: self.ua.random, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Upgrade-Insecure-Requests: 1, } if referer: headers[Referer] referer return headers def random_delay(self, min_sec1, max_sec3): 随机延迟模拟人类行为 time.sleep(random.uniform(min_sec, max_sec)) def fetch_with_retry(self, url, max_retries3, **kwargs): 带重试的请求 for attempt in range(max_retries): try: headers self.get_headers() response self.session.get(url, headersheaders, timeout30, **kwargs) if response.status_code 200: return response elif response.status_code 403: print(f被拒绝访问等待后重试... (尝试 {attempt1}/{max_retries})) time.sleep(random.uniform(5, 10)) else: print(f状态码 {response.status_code}重试中...) except Exception as e: print(f请求错误{e}重试中...) time.sleep(2 ** attempt) return None7.2 代理IP支持对于需要频繁请求的场景使用代理IP可以有效避免IP被封禁pythonclass ProxyManager: def __init__(self, proxy_listNone): self.proxies proxy_list or [] self.current_index 0 self.lock threading.Lock() def add_proxy(self, proxy): 添加代理 self.proxies.append(proxy) def get_proxy(self): 获取下一个代理轮询 with self.lock: if not self.proxies: return None proxy self.proxies[self.current_index] self.current_index (self.current_index 1) % len(self.proxies) return proxy def get_proxy_dict(self): 获取requests格式的代理字典 proxy self.get_proxy() if not proxy: return None return { http: proxy, https: proxy }7.3 数据库存储与任务管理对于大量下载任务使用数据库管理更加高效pythonimport sqlite3 import json from datetime import datetime class DownloadDatabase: def __init__(self, db_pathdownloads.db): self.conn sqlite3.connect(db_path) self.create_tables() def create_tables(self): 创建数据表 cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL, filename TEXT, save_path TEXT, status TEXT DEFAULT pending, file_size INTEGER DEFAULT 0, downloaded_size INTEGER DEFAULT 0, retry_count INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, completed_at TIMESTAMP, error_msg TEXT ) ) cursor.execute( CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def add_task(self, url, filenameNone): 添加下载任务 cursor self.conn.cursor() cursor.execute( INSERT INTO tasks (url, filename, status) VALUES (?, ?, ?), (url, filename, pending) ) self.conn.commit() return cursor.lastrowid def update_task_status(self, task_id, status, **kwargs): 更新任务状态 fields [] values [] for key, value in kwargs.items(): fields.append(f{key} ?) values.append(value) values.append(task_id) cursor self.conn.cursor() query fUPDATE tasks SET status ?, {, .join(fields)} WHERE id ? cursor.execute(query, [status] values) self.conn.commit() def get_pending_tasks(self, limit10): 获取待处理任务 cursor self.conn.cursor() cursor.execute( SELECT id, url, filename FROM tasks WHERE status pending LIMIT ?, (limit,) ) return cursor.fetchall() def close(self): 关闭数据库连接 self.conn.close()7.4 日志系统完善的日志记录对于调试和监控至关重要pythonimport logging import sys from logging.handlers import RotatingFileHandler def setup_logger(name, log_levellogging.INFO): 配置日志系统 logger logging.getLogger(name) logger.setLevel(log_level) # 控制台输出 console_handler logging.StreamHandler(sys.stdout) console_handler.setLevel(log_level) console_format logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s, datefmt%Y-%m-%d %H:%M:%S ) console_handler.setFormatter(console_format) # 文件输出带轮转 file_handler RotatingFileHandler( flogs/{name}.log, maxBytes10*1024*1024, # 10MB backupCount5, encodingutf-8 ) file_handler.setLevel(log_level) file_format logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s ) file_handler.setFormatter(file_format) logger.addHandler(console_handler) logger.addHandler(file_handler) return logger八、完整实战案例8.1 案例从学术论文网站批量下载PDF现在我们结合以上所有功能完成一个完整的实战案例python#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import sys import argparse from typing import List, Dict import requests from bs4 import BeautifulSoup from urllib.parse import urljoin from downloader import resume_download from parser import extract_pdf_links, filter_pdf_links from utils import setup_logger, AntiScrapingManager, ProgressCallback from config import Config logger setup_logger(pdf_downloader) class AcademicPDFDownloader: 学术论文PDF批量下载器 def __init__(self, configNone): self.config config or Config() self.anti_scraping AntiScrapingManager() self.session requests.Session() def search_papers(self, keyword: str, max_pages: int 5) - List[Dict]: 根据关键词搜索论文返回论文信息列表 这里以arXiv为例实际使用需根据目标网站调整 papers [] base_url https://arxiv.org/search/ for page in range(1, max_pages 1): params { query: keyword, searchtype: all, source: header, start: (page - 1) * 25 } try: response self.anti_scraping.fetch_with_retry( base_url, paramsparams ) if not response: continue soup BeautifulSoup(response.text, html.parser) # 解析论文列表 for result in soup.select(.arxiv-result): title_elem result.select_one(.title a) pdf_elem result.select_one(.pdf-link a) if pdf_elem and title_elem: papers.append({ title: title_elem.text.strip(), pdf_url: urljoin(base_url, pdf_elem.get(href)), abs_url: title_elem.get(href) }) logger.info(f第 {page} 页找到 {len(papers)} 篇论文) self.anti_scraping.random_delay(1, 2) except Exception as e: logger.error(f搜索第 {page} 页失败{e}) continue return papers def download_papers(self, papers: List[Dict], save_dir: str downloads): 批量下载论文PDF os.makedirs(save_dir, exist_okTrue) success_count 0 total len(papers) for i, paper in enumerate(papers, 1): title paper[title] pdf_url paper[pdf_url] # 清理文件名 filename self._sanitize_filename(title) .pdf save_path os.path.join(save_dir, filename) logger.info(f[{i}/{total}] 下载{filename}) print(f\n正在下载 [{i}/{total}]{title[:50]}...) try: success resume_download(pdf_url, save_path) if success: success_count 1 logger.info(f✅ 下载成功{filename}) else: logger.warning(f❌ 下载失败{filename}) except Exception as e: logger.error(f下载异常{filename}{e}) logger.info(f下载完成成功 {success_count}/{total}) return success_count, total - success_count def _sanitize_filename(self, filename: str) - str: 清理文件名移除非法字符 # 移除或替换Windows/Linux文件名中不允许的字符 illegal_chars :/\\|?* for char in illegal_chars: filename filename.replace(char, _) # 限制文件名长度 if len(filename) 200: filename filename[:200] return filename def main(): parser argparse.ArgumentParser(description学术PDF批量下载工具) parser.add_argument(-k, --keyword, requiredTrue, help搜索关键词) parser.add_argument(-p, --pages, typeint, default3, help搜索页数) parser.add_argument(-o, --output, defaultdownloads, help保存目录) parser.add_argument(-w, --workers, typeint, default3, help并发下载数) parser.add_argument(-r, --resume, actionstore_true, help启用断点续传) args parser.parse_args() # 创建下载器 downloader AcademicPDFDownloader() try: # 搜索论文 print(f 搜索关键词{args.keyword}) papers downloader.search_papers(args.keyword, args.pages) print(f 找到 {len(papers)} 篇论文) if not papers: print(未找到任何论文请检查关键词或网络连接) return # 确认下载 print(f\n准备下载 {len(papers)} 篇论文到 {args.output}) confirm input(是否继续(y/n): ) if confirm.lower() ! y: print(已取消下载) return # 下载论文 success, failed downloader.download_papers(papers, args.output) print(f\n 下载统计成功 {success} 篇失败 {failed} 篇) except KeyboardInterrupt: print(\n⏹ 用户中断下载) except Exception as e: print(f❌ 程序异常{e}) logger.exception(程序异常退出) if __name__ __main__: main()8.2 配置文件管理将可配置参数放在独立的配置文件中python# config.py import os from dataclasses import dataclass from typing import List, Optional dataclass class Config: 全局配置类 # 下载配置 max_retries: int 3 timeout: int 30 chunk_size: int 8192 max_workers: int 3 # 代理配置 proxy_list: List[str] None # 请求配置 user_agents: List[str] None delay_min: float 1.0 delay_max: float 3.0 # 文件配置 download_dir: str downloads temp_suffix: str .temp max_filename_length: int 200 # 日志配置 log_level: str INFO log_dir: str logs log_max_bytes: int 10 * 1024 * 1024 log_backup_count: int 5 # 数据库配置 db_path: str downloads.db def __post_init__(self): # 创建必要的目录 os.makedirs(self.download_dir, exist_okTrue) os.makedirs(self.log_dir, exist_okTrue)九、性能优化与最佳实践9.1 并发下载的性能考量并发下载可以显著提升下载效率但需要注意以下几点合理设置并发数并发数过高可能导致网络拥塞或IP被封建议根据网络状况和目标网站限制调整连接复用使用Session对象复用TCP连接减少握手开销异步IO对于I/O密集型任务可考虑使用asyncio aiohttp实现更高性能python# 使用aiohttp实现异步下载高级优化 import aiohttp import asyncio import aiofiles async def async_download(session, url, save_path, semaphore): 异步下载单个文件 async with semaphore: try: async with session.get(url) as response: if response.status 200: total_size int(response.headers.get(content-length, 0)) downloaded 0 async with aiofiles.open(save_path, wb) as f: async for chunk in response.content.iter_chunked(8192): await f.write(chunk) downloaded len(chunk) return True, save_path else: return False, f状态码{response.status} except Exception as e: return False, str(e)9.2 内存优化下载大文件时内存管理尤为重要始终使用流式下载streamTrue设置合适的chunk_size通常8KB-64KB及时释放不需要的响应对象使用生成器或迭代器处理大量URL9.3 错误处理与重试策略完善的错误处理机制是程序健壮性的保证pythonclass RetryStrategy: 重试策略 staticmethod def exponential_backoff(attempt, base_delay1, max_delay60): 指数退避 delay min(base_delay * (2 ** attempt), max_delay) return delay random.uniform(0, 0.5) # 增加随机抖动 staticmethod def should_retry(error, attempt, max_retries): 判断是否应该重试 if attempt max_retries: return False # 可重试的错误类型 retryable_errors ( requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.ChunkedEncodingError, ) if isinstance(error, retryable_errors): return True # 特定状态码重试 if hasattr(error, response) and hasattr(error.response, status_code): if error.response.status_code in [429, 503, 504]: return True return False9.4 测试用例编写单元测试确保代码质量python# tests/test_downloader.py import unittest import os import tempfile from unittest.mock import Mock, patch from src.downloader import resume_download class TestDownloader(unittest.TestCase): def setUp(self): self.temp_dir tempfile.mkdtemp() self.test_url https://example.com/test.pdf def tearDown(self): # 清理临时文件 import shutil shutil.rmtree(self.temp_dir) patch(requests.get) def test_resume_download_success(self, mock_get): 测试断点续传成功场景 # 模拟响应 mock_response Mock() mock_response.status_code 206 mock_response.headers {content-length: 1000} mock_response.iter_content.return_value [bx * 1000] mock_get.return_value mock_response save_path os.path.join(self.temp_dir, test.pdf) result resume_download(self.test_url, save_path) self.assertTrue(result) self.assertTrue(os.path.exists(save_path)) self.assertEqual(os.path.getsize(save_path), 1000) patch(requests.get) def test_resume_download_with_existing(self, mock_get): 测试已有文件的续传 # 创建部分文件 temp_path os.path.join(self.temp_dir, test.pdf.temp) with open(temp_path, wb) as f: f.write(bx * 500) # 模拟响应 mock_response Mock() mock_response.status_code 206 mock_response.headers {content-length: 500} mock_response.iter_content.return_value [bx * 500] mock_get.return_value mock_response save_path os.path.join(self.temp_dir, test.pdf) result resume_download(self.test_url, save_path) self.assertTrue(result) # 验证合并后的文件 self.assertTrue(os.path.exists(save_path)) self.assertEqual(os.path.getsize(save_path), 1000) if __name__ __main__: unittest.main()十、常见问题与解决方案10.1 SSL证书验证问题在请求某些网站时可能会遇到SSL证书验证失败的问题python# 解决方案1关闭验证不推荐用于生产环境 response requests.get(url, verifyFalse) # 解决方案2指定自定义证书 response requests.get(url, verify/path/to/cert.pem) # 解决方案3忽略证书错误配合警告抑制 import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) response requests.get(url, verifyFalse)10.2 中文文件名乱码处理中文文件名时需要正确的编码pythonfrom urllib.parse import unquote def decode_filename(filename): 解码URL编码的文件名 try: # 尝试多种编码 decoded unquote(filename, encodingutf-8) return decoded except UnicodeDecodeError: try: return unquote(filename, encodinggbk) except: return filename10.3 大文件内存不足当下载超大文件如几十GB时注意以下几点始终使用流式下载不要使用response.content避免将整个文件加载到内存中进行校验考虑使用mmap进行内存映射定期保存进度防止意外中断10.4 服务器限制文件访问有些服务器会限制PDF文件的直接访问需要处理Cookie或登录状态pythonclass AuthenticatedDownloader: def __init__(self, login_url, username, password): self.session requests.Session() self.login(login_url, username, password) def login(self, login_url, username, password): 登录获取Cookie payload {username: username, password: password} response self.session.post(login_url, datapayload) response.raise_for_status() # 验证登录状态 if sessionid in self.session.cookies: print(登录成功) else: print(登录失败) def download(self, url, save_path): 使用已认证的session下载 response self.session.get(url, streamTrue) # 后续下载逻辑...十一、总结与展望本文详细介绍了如何使用Python开发一个功能完善的PDF文档批量下载器涵盖了从基础功能到高级优化的各个方面。主要技术要点包括requests库的核心用法流式下载、重定向处理、请求头定制文件操作二进制文件的流式写入、临时文件管理进度显示使用tqdm实现美观的进度条断点续传基于HTTP Range协议实现大文件续传批量管理并发下载、任务队列、数据库记录反爬策略User-Agent轮换、延迟控制、代理IP错误处理重试机制、异常捕获、日志记录未来改进方向Web界面使用Flask或Django开发Web管理界面分布式下载使用Redis或消息队列实现多节点协同下载智能识别使用机器学习自动分类下载的PDF文档OCR集成对扫描版PDF进行文字识别建立全文搜索云存储集成直接上传到阿里云OSS或AWS S3浏览器扩展开发Chrome/Edge扩展一键导出PDF链接