当前位置：首页 > news >正文

Python爬虫实战：利用最新技术高效抓取电子书资源

news 2026/5/12 6:54:04

引言

在数字时代，电子书已成为获取知识的重要途径。然而，手动从各个网站收集电子书既耗时又低效。本文将详细介绍如何使用Python最新爬虫技术，构建一个高效、稳定的电子书资源下载工具。我们将涵盖异步请求、反爬对抗、智能解析等前沿技术，并提供完整的代码实现。

技术栈概览

Python 3.9+- 最新Python版本支持
aiohttp/asyncio- 异步HTTP请求处理
Playwright- 现代浏览器自动化工具
BeautifulSoup4/lxml- HTML解析
Scrapy框架- 高级爬虫框架
Redis- 分布式任务队列和缓存
代理IP池- 反反爬虫策略

项目架构设计

1. 核心模块划分

text

ebook-spider/ ├── spiders/ # 爬虫核心逻辑 ├── middleware/ # 中间件（代理、UA轮换等） ├── pipeline/ # 数据处理管道 ├── utils/ # 工具函数 ├── config.py # 配置文件 └── main.py # 入口文件

2. 完整代码实现

python

""" 电子书资源智能爬虫系统 支持异步并发、动态渲染页面处理、反反爬虫机制 """ import asyncio import aiohttp import logging from typing import Optional, List, Dict, Any from dataclasses import dataclass from urllib.parse import urljoin, urlparse import json import hashlib from datetime import datetime from playwright.async_api import async_playwright import redis from bs4 import BeautifulSoup import re # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @dataclass class EbookInfo: """电子书信息数据类""" title: str author: str publisher: str publish_date: str isbn: str format: str # PDF, EPUB, MOBI等 language: str file_size: str download_url: str source_url: str description: str = "" categories: List[str] = None cover_url: str = "" def __post_init__(self): if self.categories is None: self.categories = [] def to_dict(self) -> Dict[str, Any]: return { 'title': self.title, 'author': self.author, 'publisher': self.publisher, 'publish_date': self.publish_date, 'isbn': self.isbn, 'format': self.format, 'language': self.language, 'file_size': self.file_size, 'download_url': self.download_url, 'source_url': self.source_url, 'description': self.description, 'categories': self.categories, 'cover_url': self.cover_url, 'crawl_time': datetime.now().isoformat() } class AsyncRequestClient: """异步HTTP请求客户端，集成代理和重试机制""" def __init__(self, max_retries: int = 3, timeout: int = 30): self.max_retries = max_retries self.timeout = aiohttp.ClientTimeout(total=timeout) self.session: Optional[aiohttp.ClientSession] = None self.proxy_pool = [] # 代理IP池 self.user_agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36', ] async def __aenter__(self): self.session = aiohttp.ClientSession(timeout=self.timeout) return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() def _get_random_headers(self) -> Dict[str, str]: """生成随机请求头""" import random return { 'User-Agent': random.choice(self.user_agents), 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', 'Cache-Control': 'max-age=0' } async def fetch(self, url: str, method: str = 'GET', **kwargs) -> Optional[str]: """带重试机制的异步请求""" for attempt in range(self.max_retries): try: headers = self._get_random_headers() # 如果有代理池，随机选择代理 if self.proxy_pool: proxy = random.choice(self.proxy_pool) kwargs['proxy'] = proxy async with self.session.request( method=method, url=url, headers=headers, **kwargs ) as response: if response.status == 200: return await response.text() elif response.status == 429: # 频率限制 wait_time = 2 ** attempt # 指数退避 logger.warning(f"Rate limited, waiting {wait_time}s") await asyncio.sleep(wait_time) else: logger.error(f"HTTP {response.status} for {url}") except (aiohttp.ClientError, asyncio.TimeoutError) as e: logger.warning(f"Attempt {attempt + 1} failed for {url}: {e}") if attempt == self.max_retries - 1: logger.error(f"Failed to fetch {url} after {self.max_retries} attempts") return None await asyncio.sleep(1) # 基础延迟 return None class DynamicPageRenderer: """处理JavaScript动态渲染的页面""" async def render_page(self, url: str, wait_for_selector: str = None) -> Optional[str]: """使用Playwright渲染动态页面""" async with async_playwright() as p: # 启动浏览器（可配置为无头模式） browser = await p.chromium.launch(headless=True) context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' ) page = await context.new_page() try: await page.goto(url, wait_until='networkidle') # 等待特定元素加载（如果指定） if wait_for_selector: await page.wait_for_selector(wait_for_selector, timeout=10000) # 滚动页面以加载懒加载内容 await self._auto_scroll(page) # 获取页面内容 content = await page.content() return content except Exception as e: logger.error(f"Error rendering page {url}: {e}") return None finally: await browser.close() async def _auto_scroll(self, page): """自动滚动页面以触发懒加载""" await page.evaluate(""" async () => { await new Promise((resolve) => { let totalHeight = 0; const distance = 100; const timer = setInterval(() => { const scrollHeight = document.body.scrollHeight; window.scrollBy(0, distance); totalHeight += distance; if(totalHeight >= scrollHeight){ clearInterval(timer); resolve(); } }, 100); }); } """) class EbookSpider: """电子书爬虫主类""" def __init__(self, base_urls: List[str], max_concurrent: int = 5): self.base_urls = base_urls self.max_concurrent = max_concurrent self.request_client = AsyncRequestClient() self.renderer = DynamicPageRenderer() self.redis_client = redis.Redis(host='localhost', port=6379, db=0) self.visited_urls = set() self.ebooks_found = [] # 文件扩展名模式 self.file_patterns = { 'pdf': r'\.pdf($|\?)', 'epub': r'\.epub($|\?)', 'mobi': r'\.mobi($|\?)', 'azw3': r'\.azw3($|\?)' } def _extract_ebook_info(self, soup: BeautifulSoup, url: str) -> Optional[EbookInfo]: """从页面提取电子书信息（需根据具体网站结构调整）""" try: # 尝试多种选择器提取标题 title_selectors = [ 'h1[class*="title"]', 'h1[class*="book"]', '.book-title', '#bookTitle', 'title' ] title = None for selector in title_selectors: element = soup.select_one(selector) if element: title = element.get_text(strip=True) break if not title: return None # 提取作者信息 author = "Unknown" author_selectors = [ 'a[class*="author"]', '.book-author', '[itemprop="author"]', 'span:contains("Author") + a' ] for selector in author_selectors: element = soup.select_one(selector) if element: author = element.get_text(strip=True) break # 查找下载链接 download_url = None for format_name, pattern in self.file_patterns.items(): links = soup.find_all('a', href=re.compile(pattern, re.I)) if links: download_url = urljoin(url, links[0]['href']) file_format = format_name.upper() break if not download_url: return None # 构建EbookInfo对象 ebook = EbookInfo( title=title, author=author, publisher=self._extract_metadata(soup, 'publisher'), publish_date=self._extract_metadata(soup, 'publish_date'), isbn=self._extract_metadata(soup, 'isbn'), format=file_format, language=self._extract_metadata(soup, 'language', 'English'), file_size=self._extract_file_size(soup, download_url), download_url=download_url, source_url=url, description=self._extract_description(soup), categories=self._extract_categories(soup), cover_url=self._extract_cover_url(soup, url) ) return ebook except Exception as e: logger.error(f"Error extracting ebook info: {e}") return None def _extract_metadata(self, soup: BeautifulSoup, field: str, default: str = "") -> str: """提取特定元数据字段""" # 实现根据网站结构提取具体字段的逻辑 return default def _extract_file_size(self, soup: BeautifulSoup, download_url: str) -> str: """提取文件大小信息""" # 可以从链接文本或单独的元素中提取 return "Unknown" def _extract_description(self, soup: BeautifulSoup) -> str: """提取描述信息""" selectors = [ 'div[class*="description"]', '#bookDescription', '.summary', '[itemprop="description"]' ] for selector in selectors: element = soup.select_one(selector) if element: return element.get_text(strip=True)[:500] # 限制长度 return "" def _extract_categories(self, soup: BeautifulSoup) -> List[str]: """提取分类标签""" categories = [] category_links = soup.select('a[class*="category"], a[class*="tag"]') for link in category_links[:5]: # 最多取5个 categories.append(link.get_text(strip=True)) return categories def _extract_cover_url(self, soup: BeautifulSoup, base_url: str) -> str: """提取封面图片URL""" img_selectors = [ 'img[class*="cover"]', '.book-cover img', '[itemprop="image"]', 'meta[property="og:image"]' ] for selector in img_selectors: element = soup.select_one(selector) if element: src = element.get('src') or element.get('content') if src: return urljoin(base_url, src) return "" async def crawl_page(self, url: str) -> List[EbookInfo]: """爬取单个页面""" # 检查是否已访问 url_hash = hashlib.md5(url.encode()).hexdigest() if self.redis_client.exists(f'visited:{url_hash}'): return [] logger.info(f"Crawling: {url}") # 首先尝试直接请求 html = await self.request_client.fetch(url) # 如果失败或需要动态渲染，使用Playwright if not html or '动态内容检测' in html: # 这里可以添加更智能的检测 html = await self.renderer.render_page(url, wait_for_selector='.book-info') if not html: return [] # 解析HTML soup = BeautifulSoup(html, 'lxml') # 提取电子书信息 ebooks = [] main_ebook = self._extract_ebook_info(soup, url) if main_ebook: ebooks.append(main_ebook) # 查找页面中的其他电子书链接 book_links = soup.select('a[href*="book"], a[href*="ebook"]') for link in book_links[:10]: # 限制数量防止过度爬取 book_url = urljoin(url, link.get('href')) if book_url not in self.visited_urls: self.visited_urls.add(book_url) # 可以递归爬取，这里简单处理 # 标记为已访问 self.redis_client.setex(f'visited:{url_hash}', 86400, '1') # 24小时过期 return ebooks async def crawl_all(self): """并发爬取所有基础URL""" semaphore = asyncio.Semaphore(self.max_concurrent) async def crawl_with_semaphore(url): async with semaphore: return await self.crawl_page(url) tasks = [crawl_with_semaphore(url) for url in self.base_urls] results = await asyncio.gather(*tasks, return_exceptions=True) # 合并结果 for result in results: if isinstance(result, list): self.ebooks_found.extend(result) logger.info(f"Found {len(self.ebooks_found)} ebooks") def save_results(self, format: str = 'json', filename: str = 'ebooks'): """保存爬取结果""" data = [ebook.to_dict() for ebook in self.ebooks_found] if format == 'json': with open(f'{filename}.json', 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) elif format == 'csv': import pandas as pd df = pd.DataFrame(data) df.to_csv(f'{filename}.csv', index=False, encoding='utf-8-sig') logger.info(f"Results saved to {filename}.{format}") class DownloadManager: """电子书下载管理器""" @staticmethod async def download_ebook(ebook: EbookInfo, save_dir: str = './downloads'): """异步下载电子书文件""" import os from aiofiles import open as aioopen os.makedirs(save_dir, exist_ok=True) # 生成安全文件名 safe_title = re.sub(r'[^\w\s-]', '', ebook.title) filename = f"{safe_title}_{ebook.isbn or 'unknown'}.{ebook.format.lower()}" filepath = os.path.join(save_dir, filename) async with aiohttp.ClientSession() as session: try: async with session.get(ebook.download_url) as response: if response.status == 200: async with aioopen(filepath, 'wb') as f: while True: chunk = await response.content.read(8192) if not chunk: break await f.write(chunk) logger.info(f"Downloaded: {filename}") return filepath except Exception as e: logger.error(f"Failed to download {ebook.title}: {e}") return None async def main(): """主函数""" # 配置目标网站（示例） target_sites = [ 'https://example-ebooks-site.com/category/programming', 'https://another-ebook-site.org/latest', # 添加更多合法、允许爬取的网站 ] # 初始化爬虫 spider = EbookSpider(target_sites, max_concurrent=3) # 开始爬取 logger.info("Starting ebook spider...") await spider.crawl_all() # 保存结果 spider.save_results(format='json', filename='ebooks_collection') # 可选：下载电子书文件 downloader = DownloadManager() download_tasks = [] for ebook in spider.ebooks_found[:5]: # 限制下载数量 task = downloader.download_ebook(ebook) download_tasks.append(task) # 并发下载 await asyncio.gather(*download_tasks) logger.info("Spider finished successfully!") if __name__ == "__main__": # 运行异步主函数 asyncio.run(main())

高级功能扩展

1. 分布式爬虫实现

python

import celery from celery import Celery # 配置Celery分布式任务队列 app = Celery('ebook_spider', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0') @app.task def distributed_crawl_task(url): """分布式爬虫任务""" # 实现分布式爬取逻辑 pass

2. 智能反反爬虫策略

python

class AntiAntiSpider: """智能反反爬虫系统""" def __init__(self): self.behavior_pattern = { 'request_delay': (1, 3), # 随机延迟范围 'mouse_movements': True, # 模拟鼠标移动 'random_scroll': True, # 随机滚动 'human_typing': True # 模拟人类输入 } async def human_like_behavior(self, page): """模拟人类浏览器行为""" # 实现复杂的人类行为模拟 pass

3. 数据去重与增量爬取

python

class DeduplicationEngine: """基于内容哈希的数据去重引擎""" def __init__(self): self.bloom_filter = BloomFilter(capacity=1000000, error_rate=0.001) def is_duplicate(self, content: str) -> bool: """检查内容是否重复""" content_hash = hashlib.sha256(content.encode()).hexdigest() return content_hash in self.bloom_filter

最佳实践与注意事项

1. 法律与道德考量

尊重robots.txt：遵守网站的爬取政策
限制请求频率：避免对目标网站造成压力
仅爬取公开数据：不访问需要认证的私人内容
遵守版权法：仅用于个人学习研究

2. 性能优化技巧

使用连接池复用HTTP连接
实现增量爬取，避免重复工作
合理设置并发数，避免被封锁
使用压缩传输减少带宽消耗

3. 错误处理与监控

实现完善的日志系统
设置警报机制监控爬虫状态
添加健康检查端点
实现自动恢复机制

4. 数据存储方案

python

# 多格式存储支持 STORAGE_BACKENDS = { 'mongodb': MongoDBStorage, 'postgresql': PostgreSQLStorage, 'elasticsearch': ElasticsearchStorage, 's3': S3Storage }

部署与维护

Docker部署配置

dockerfile

FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["python", "main.py"]

配置管理

python

# config.yaml spider: max_concurrent: 5 request_timeout: 30 retry_times: 3 storage: type: "mongodb" uri: "mongodb://localhost:27017" proxy: enabled: true pool_size: 10