# Python爬虫核心架构组件详解 Python爬虫架构主要由五个核心组件构成，每个组件承担着不同的功能，共同协作完成网络数据的抓取、解析和处理任务[ref_1]。 ## 核心架构组件功能表 | 组件名称 | 主要功能 | 实现方式/工具 | 关键作用 | |---------|---------|--------------|---------| | **调度器** | 协调各组件工作流程 | 自定义调度逻辑 | 相当于爬虫的CPU，控制整体执行流程[ref_1] | | **URL管理器** | 管理待抓取和已抓取URL | 内存、数据库、缓存数据库 | 防止重复抓取和循环抓取[ref_1] | | **网页下载器** | 下载网页内容 | urllib、requests等库 | 将URL转换为网页字符串[ref_1] | | **网页解析器** | 提取有用信息 | 正则表达式、BeautifulSoup、lxml等 | 解析网页结构，提取目标数据[ref_1] | | **应用程序** | 处理和使用抓取数据 | 数据库、文件、API等 | 存储和应用提取的有价值数据[ref_1] | ## 各组件详细功能解析 ### 1. 调度器 (Scheduler) 调度器是整个爬虫系统的"大脑"，负责控制数据流的传递和组件间的协调工作[ref_1]。 ```python class Scheduler: def __init__(self): self.url_manager = URLManager() self.downloader = Downloader() self.parser = Parser() self.data_processor = DataProcessor() def schedule(self, start_urls): """调度整个爬虫流程""" # 1. 初始化URL管理器 self.url_manager.add_new_urls(start_urls) while self.url_manager.has_new_url(): # 2. 从URL管理器获取待抓取URL url = self.url_manager.get_new_url() # 3. 调用下载器下载网页 html_content = self.downloader.download(url) # 4. 调用解析器解析网页 new_urls, data = self.parser.parse(html_content, url) # 5. 将新URL添加到URL管理器 self.url_manager.add_new_urls(new_urls) # 6. 处理提取的数据 self.data_processor.process(data) ``` ### 2. URL管理器 (URL Manager) URL管理器负责维护URL的状态，确保爬虫高效且不重复地工作[ref_1]。 ```python class URLManager: def __init__(self): self.new_urls = set() # 待抓取URL集合 self.old_urls = set() # 已抓取URL集合 def add_new_url(self, url): """添加单个新URL""" if url is None or url == "": return if url not in self.new_urls and url not in self.old_urls: self.new_urls.add(url) def add_new_urls(self, urls): """批量添加新URL""" if urls is None or len(urls) == 0: return for url in urls: self.add_new_url(url) def has_new_url(self): """判断是否有待抓取URL""" return len(self.new_urls) > 0 def get_new_url(self): """获取一个待抓取URL""" new_url = self.new_urls.pop() self.old_urls.add(new_url) return new_url ``` ### 3. 网页下载器 (Web Downloader) 下载器负责通过网络请求获取网页内容，支持各种HTTP请求和处理[ref_1]。 ```python import requests from urllib.request import urlopen import time class Downloader: def __init__(self, delay=1): self.delay = delay # 请求延迟，避免被封 self.session = requests.Session() # 设置请求头，模拟浏览器行为 self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } def download_with_urllib(self, url): """使用urllib下载网页""" try: # 创建请求对象并设置headers req = urllib.request.Request(url, headers=self.headers) response = urllib.request.urlopen(req, timeout=10) html = response.read().decode('utf-8') return html except Exception as e: print(f"下载失败: {e}") return None def download_with_requests(self, url): """使用requests库下载网页""" try: time.sleep(self.delay) # 延迟请求 response = self.session.get(url, headers=self.headers, timeout=10) response.encoding = 'utf-8' return response.text except Exception as e: print(f"下载失败: {e}") return None ``` ### 4. 网页解析器 (Web Parser) 解析器负责从网页HTML中提取结构化数据，支持多种解析方式[ref_1]。 ```python from bs4 import BeautifulSoup import re import lxml.html class Parser: def __init__(self): self.patterns = { 'title': re.compile(r'<title>(.*?)</title>'), 'links': re.compile(r'href="(http.*?)"') } def parse_with_bs4(self, html, base_url): """使用BeautifulSoup解析网页""" soup = BeautifulSoup(html, 'html.parser') # 提取数据 data = { 'url': base_url, 'title': soup.title.string if soup.title else '', 'content': soup.get_text(), 'links': [] } # 提取所有链接 new_urls = [] for link in soup.find_all('a', href=True): href = link['href'] # 处理相对URL if href.startswith('/'): full_url = base_url + href elif href.startswith('http'): full_url = href else: continue new_urls.append(full_url) data['links'].append({'text': link.get_text(), 'url': full_url}) return new_urls, data def parse_with_regex(self, html, base_url): """使用正则表达式解析网页""" data = {} new_urls = [] # 提取标题 title_match = self.patterns['title'].search(html) data['title'] = title_match.group(1) if title_match else '' # 提取链接 link_matches = self.patterns['links'].findall(html) new_urls.extend(link_matches) return new_urls, data ``` ### 5. 应用程序 (Application) 应用程序负责处理和存储提取的数据，可以根据需求进行定制化开发[ref_1]。 ```python import json import csv import sqlite3 class DataProcessor: def __init__(self): self.data_list = [] def process(self, data): """处理提取的数据""" if data: self.data_list.append(data) self.save_to_json(data) self.save_to_database(data) def save_to_json(self, data): """保存数据到JSON文件""" with open('crawled_data.json', 'a', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) f.write('\n') def save_to_database(self, data): """保存数据到SQLite数据库""" conn = sqlite3.connect('crawler.db') cursor = conn.cursor() # 创建表（如果不存在） cursor.execute(''' CREATE TABLE IF NOT EXISTS pages ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT UNIQUE, title TEXT, content TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ''') # 插入数据 cursor.execute(''' INSERT OR IGNORE INTO pages (url, title, content) VALUES (?, ?, ?) ''', (data['url'], data.get('title', ''), data.get('content', ''))) conn.commit() conn.close() ``` ## 实际应用场景示例 ### 完整爬虫系统集成 ```python def main(): """完整的爬虫系统示例""" # 初始化所有组件 scheduler = Scheduler() url_manager = URLManager() downloader = Downloader(delay=2) parser = Parser() data_processor = DataProcessor() # 设置起始URL start_urls = ['https://example.com'] # 开始爬取 url_manager.add_new_urls(start_urls) max_pages = 100 # 限制爬取页面数量 crawled_count = 0 while url_manager.has_new_url() and crawled_count < max_pages: # 获取URL current_url = url_manager.get_new_url() print(f"爬取: {current_url}") # 下载网页 html = downloader.download_with_requests(current_url) if not html: continue # 解析网页 new_urls, extracted_data = parser.parse_with_bs4(html, current_url) # 更新URL管理器 url_manager.add_new_urls(new_urls) # 处理数据 data_processor.process(extracted_data) crawled_count += 1 print(f"已完成 {crawled_count} 个页面的爬取") if __name__ == "__main__": main() ``` ## 组件协同工作流程 1. **初始化阶段**：调度器初始化所有组件，设置起始URL[ref_1] 2. **URL管理**：URL管理器维护待抓取和已抓取URL队列[ref_1] 3. **下载循环**：调度器从URL管理器获取URL，调用下载器获取网页内容[ref_1] 4. **解析处理**：下载的内容传递给解析器提取数据和新的URL[ref_1] 5. **数据应用**：提取的数据由应用程序进行存储或进一步处理[ref_1] 6. **URL更新**：新发现的URL被添加回URL管理器，循环继续[ref_1] 这种模块化的架构设计使得Python爬虫具有高度的可扩展性和可维护性，每个组件都可以独立优化和替换，适应不同的爬取需求和技术栈[ref_1]。