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

维基百科分类页面爬虫实战：递归获取所有页面标题

news 2026/6/12 1:23:29

一、项目背景与目标

在数据科学和自然语言处理领域，维基百科是一个极为宝贵的数据源。它包含了数以百万计的结构化知识条目，覆盖了几乎所有学科领域。在实际应用中，我们经常需要获取某个主题分类下的所有页面标题，例如“机器学习”分类下的所有算法页面，“人工智能”分类下的所有相关概念，或者“中国城市”分类下的所有城市条目。这些页面标题可以用于构建知识图谱、训练词向量模型、进行主题建模，或者作为更大规模爬虫的种子URL。

本文将带您从零开始，编写一个完整的Python爬虫，实现以下目标：

从维基百科的某个指定分类页面开始（例如https://en.wikipedia.org/wiki/Category:Machine_learning）
递归地爬取该分类下的所有子分类页面
提取每个分类页面中直接包含的普通文章页面标题
避免重复爬取和死循环（循环引用的分类关系）
支持断点续爬和结果保存

本文注重实战，代码全部可运行，并会详细解释每个技术点的设计思路和实现细节。

二、技术选型与准备工作

2.1 为什么选择维基百科？

维基百科具有以下特点，使其成为爬虫练习的理想目标：

结构清晰：页面URL模式固定，分类页面有明确的标识。
友好的反爬策略：允许爬虫访问，但要求遵守robots.txt和合理设置请求间隔。
丰富的链接关系：分类之间通过超链接关联，天然适合递归爬取。
数据价值高：每个页面都是高质量的结构化文本。

2.2 核心库选择

我们将使用以下Python库：

库名	用途
`requests`	发送HTTP请求，获取HTML内容
`beautifulsoup4`	解析HTML，提取链接和文本
`urllib.parse`	处理相对URL和分类路径解析
`time`	控制请求间隔，避免被封
`json`	保存爬取结果和进度
`logging`	记录日志，便于调试和监控
`dataclasses`	定义数据结构

2.3 环境搭建

首先创建虚拟环境并安装依赖：

bash

# 创建虚拟环境（可选但推荐） python -m venv wiki_crawler_env source wiki_crawler_env/bin/activate # Linux/Mac # 或 wiki_crawler_env\Scripts\activate # Windows # 安装所需库 pip install requests beautifulsoup4 lxml

lxml作为BeautifulSoup的解析器，速度比默认的html.parser更快。

三、维基百科分类页面结构分析

在编写代码前，我们需要深入理解维基百科分类页面的HTML结构。

3.1 分类页面URL规则

普通页面：https://en.wikipedia.org/wiki/页面标题
分类页面：https://en.wikipedia.org/wiki/Category:分类名称

例如：

机器学习分类：https://en.wikipedia.org/wiki/Category:Machine_learning
深度学习子分类：https://en.wikipedia.org/wiki/Category:Deep_learning

3.2 页面内容组织

打开一个分类页面（例如Machine learning分类），页面主要包含三个区域：

页面说明区（页面顶部，通常包含分类的简短描述）
子分类区（Subcategories）：包含该分类下的所有子分类，每个子分类是一个链接，格式为/wiki/Category:子分类名
页面区（Pages）：包含直接属于该分类的普通文章页面，每个页面是一个链接，格式为/wiki/页面标题

此外，大型分类会分页显示，页面底部会有“下一页”链接（next page）。

3.3 HTML标签定位

通过浏览器开发者工具（F12）分析，我们可以发现：

子分类列表：位于<div id="mw-subcategories">中的<div class="CategoryTreeItem">或直接是<li>标签下的链接
页面列表：位于<div id="mw-pages">中的<li>标签下的链接
下一页链接：<a href="...">next page</a>，位于<div id="mw-pages">后的导航区域

但在不同维基百科语言版本或不同主题下，结构可能略有差异。为了健壮性，我们会使用更通用的选择器。

四、核心功能实现

4.1 请求头与会话管理

维基百科要求爬虫设置User-Agent标识自己，否则可能返回403错误。同时，为了效率，我们复用TCP连接（使用requests.Session）。

python

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session(): """创建带重试机制的requests会话""" session = requests.Session() # 设置请求头，模拟浏览器 session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', }) # 配置重试策略（网络波动时自动重试） retry_strategy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) return session

4.2 页面下载与解析

定义分类页面的解析器，提取三类信息：子分类链接、普通页面标题、下一页URL。

python

from bs4 import BeautifulSoup from urllib.parse import urljoin, urlparse import time import logging logger = logging.getLogger(__name__) class CategoryPageParser: """维基百科分类页面解析器""" BASE_URL = "https://en.wikipedia.org" @staticmethod def parse_page(html, current_url): """ 解析分类页面，返回: - subcategories: 子分类URL列表 - pages: 普通页面标题列表 - next_page_url: 下一页URL（如果有） """ soup = BeautifulSoup(html, 'lxml') subcategories = [] pages = [] next_page_url = None # 1. 提取子分类 (Subcategories) # 方法：找到 id="mw-subcategories" 的div，然后提取其中的链接 subcat_div = soup.find('div', id='mw-subcategories') if subcat_div: # 子分类可能在多种标签中，常见的是 li 或 div.CategoryTreeItem for link in subcat_div.find_all('a', href=True): href = link['href'] if href.startswith('/wiki/Category:'): full_url = urljoin(CategoryPageParser.BASE_URL, href) subcategories.append(full_url) # 备选方案：如果上面没有找到，尝试通过 class='CategoryTreeItem' 查找 if not subcategories: for item in soup.find_all('div', class_='CategoryTreeItem'): link = item.find('a', href=True) if link and link['href'].startswith('/wiki/Category:'): full_url = urljoin(CategoryPageParser.BASE_URL, link['href']) subcategories.append(full_url) # 2. 提取普通页面 (Pages) pages_div = soup.find('div', id='mw-pages') if pages_div: for link in pages_div.find_all('a', href=True): href = link['href'] # 普通页面链接以 /wiki/ 开头，但不是 /wiki/Category: if href.startswith('/wiki/') and not href.startswith('/wiki/Category:'): # 提取页面标题（URL解码，但维基百科标题通常不含特殊字符） page_title = href.replace('/wiki/', '') pages.append(page_title) # 3. 提取下一页链接 # 下一页通常在分页导航中，例如：<a href="..." class="mw-nextpage">next page</a> next_link = soup.find('a', string=lambda t: t and 'next page' in t.lower()) if not next_link: # 备选：通过 class 查找 next_link = soup.find('a', class_='mw-nextpage') if next_link and next_link.get('href'): next_page_url = urljoin(CategoryPageParser.BASE_URL, next_link['href']) # 去重（同一个分类可能在多个位置出现） subcategories = list(dict.fromkeys(subcategories)) pages = list(dict.fromkeys(pages)) return subcategories, pages, next_page_url

4.3 递归爬取与去重控制

递归爬取的核心挑战：

循环引用：分类A包含分类B，分类B也包含分类A（少见但存在）
重复爬取：同一个分类可能被多个父分类引用
深度爆炸：维基百科分类体系非常庞大，需要设置最大深度

我们采用广度优先搜索（BFS）的策略，配合visited_categories集合记录已处理的分类URL。

python

from collections import deque from dataclasses import dataclass, field from typing import Set, List, Dict import json @dataclass class CrawlState: """爬虫状态管理""" visited_categories: Set[str] = field(default_factory=set) all_page_titles: Set[str] = field(default_factory=set) category_to_pages: Dict[str, List[str]] = field(default_factory=dict) queue: deque = field(default_factory=deque) def save_checkpoint(self, filename='crawler_checkpoint.json'): """保存当前进度，用于断点续爬""" data = { 'visited_categories': list(self.visited_categories), 'all_page_titles': list(self.all_page_titles), 'category_to_pages': self.category_to_pages, 'queue': list(self.queue) } with open(filename, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) def load_checkpoint(self, filename='crawler_checkpoint.json'): """加载之前保存的进度""" try: with open(filename, 'r', encoding='utf-8') as f: data = json.load(f) self.visited_categories = set(data['visited_categories']) self.all_page_titles = set(data['all_page_titles']) self.category_to_pages = data['category_to_pages'] self.queue = deque(data['queue']) return True except FileNotFoundError: return False

4.4 主爬虫逻辑

python

class WikipediaCategoryCrawler: """维基百科分类爬虫主类""" def __init__(self, start_category_url, max_depth=3, request_delay=1.0): """ 参数: - start_category_url: 起始分类URL - max_depth: 最大递归深度（0表示只爬当前分类，1表示爬当前及直接子分类，以此类推） - request_delay: 请求间隔（秒），尊重服务器 """ self.start_url = start_category_url self.max_depth = max_depth self.request_delay = request_delay self.session = create_session() self.parser = CategoryPageParser() self.state = CrawlState() # 需要记录每个分类的深度，以便在BFS中控制深度 self.depth_map = {} # url -> depth # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) self.logger = logging.getLogger(__name__) def fetch_page(self, url): """下载页面，返回HTML文本""" try: self.logger.info(f"Fetching: {url}") response = self.session.get(url, timeout=30) response.raise_for_status() # 维基百科返回的内容编码通常是utf-8 response.encoding = 'utf-8' return response.text except requests.RequestException as e: self.logger.error(f"Failed to fetch {url}: {e}") return None def crawl(self, resume=False): """开始爬取""" if resume and self.state.load_checkpoint(): self.logger.info("Resuming from checkpoint") # 从队列中恢复时，也需要恢复depth_map（简化起见，重新从队列构建） for url in self.state.queue: # 注意：checkpoint中未保存depth，这里需要重新计算，或者保存时一起保存 pass else: # 初始化：将起始分类加入队列，深度为0 self.state.queue.append(self.start_url) self.depth_map[self.start_url] = 0 while self.state.queue: category_url = self.state.queue.popleft() current_depth = self.depth_map.get(category_url, 0) # 检查深度限制 if current_depth > self.max_depth: self.logger.info(f"Reached max depth {self.max_depth}, skipping {category_url}") continue # 检查是否已访问 if category_url in self.state.visited_categories: continue # 下载并解析 html = self.fetch_page(category_url) if not html: continue subcategories, pages, next_page_url = self.parser.parse_page(html, category_url) # 记录该分类下的页面 if pages: self.state.category_to_pages[category_url] = pages for title in pages: self.state.all_page_titles.add(title) self.logger.info(f"Found {len(pages)} pages in {category_url}") # 标记当前分类为已访问 self.state.visited_categories.add(category_url) # 处理子分类：加入队列（如果未访问且深度未超限） for subcat_url in subcategories: if subcat_url not in self.state.visited_categories: # 避免重复加入队列 if subcat_url not in self.depth_map: self.depth_map[subcat_url] = current_depth + 1 self.state.queue.append(subcat_url) self.logger.debug(f"Added subcategory: {subcat_url} (depth {current_depth+1})") # 处理分页：当前分类可能有多页，下一页中的内容仍属于同一深度 if next_page_url and next_page_url not in self.state.visited_categories: # 下一页的深度与当前分类相同 if next_page_url not in self.depth_map: self.depth_map[next_page_url] = current_depth self.state.queue.append(next_page_url) self.logger.debug(f"Added next page: {next_page_url}") # 礼貌等待，避免请求过快 time.sleep(self.request_delay) # 每处理10个分类保存一次进度（可选） if len(self.state.visited_categories) % 10 == 0: self.state.save_checkpoint() self.logger.info(f"Crawl completed. Visited {len(self.state.visited_categories)} categories, " f"found {len(self.state.all_page_titles)} unique page titles.") return self.state.all_page_titles, self.state.category_to_pages def save_results(self, titles_file='page_titles.txt', json_file='crawler_results.json'): """保存爬取结果""" # 保存为纯文本，每行一个标题 with open(titles_file, 'w', encoding='utf-8') as f: for title in sorted(self.state.all_page_titles): f.write(title + '\n') # 保存为JSON，包含详细的结构信息 results = { 'start_category': self.start_url, 'max_depth': self.max_depth, 'total_categories': len(self.state.visited_categories), 'total_pages': len(self.state.all_page_titles), 'category_to_pages': self.state.category_to_pages, 'all_titles': list(self.state.all_page_titles) } with open(json_file, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2, ensure_ascii=False) self.logger.info(f"Results saved to {titles_file} and {json_file}")

4.5 处理特殊情况

4.5.1 空分类

有些分类下既没有子分类也没有页面，解析时需优雅处理。

4.5.2 重定向分类

维基百科某些分类可能重定向到另一个分类。我们的requests默认会自动跟随重定向，最终URL会是目标分类的地址，这没有问题。但需注意，原始URL和最终URL都应标记为已访问。

4.5.3 命名空间过滤

维基百科页面有多种命名空间，如File:、Template:、Help:等。我们通常只关心普通文章（主命名空间）。在解析页面链接时，可以通过检查URL前缀来过滤：

python

def is_article_page(title_or_url): """判断是否为普通文章页面（非分类、非特殊页面）""" exclude_prefixes = ['Category:', 'File:', 'Template:', 'Help:', 'Portal:', 'Wikipedia:', 'Special:'] for prefix in exclude_prefixes: if title_or_url.startswith(prefix): return False return True

在我们的parse_page方法中，已经通过not href.startswith('/wiki/Category:')排除了子分类，但其他命名空间仍可能混入。改进如下：

python

# 在pages提取部分 if href.startswith('/wiki/') and not href.startswith('/wiki/Category:'): page_title = href.replace('/wiki/', '') # 进一步过滤其他命名空间 if is_article_page(page_title): pages.append(page_title)

五、完整代码整合

将以上模块整合成一个完整的脚本wiki_category_crawler.py：

python

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 维基百科分类爬虫 - 递归获取指定分类下所有页面标题 """ import requests import time import logging import json from collections import deque from dataclasses import dataclass, field from typing import Set, List, Dict from urllib.parse import urljoin from bs4 import BeautifulSoup from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # ========== 配置部分 ========== USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' BASE_URL = "https://en.wikipedia.org" REQUEST_DELAY = 1.0 # 请求间隔（秒） MAX_DEPTH = 2 # 最大递归深度 # ========== 辅助函数 ========== def is_article_page(title): """过滤非文章页面""" exclude = ['Category:', 'File:', 'Template:', 'Help:', 'Portal:', 'Wikipedia:', 'Special:', 'MediaWiki:', 'User:'] for prefix in exclude: if title.startswith(prefix): return False return True # ========== 会话管理 ========== def create_session(): session = requests.Session() session.headers.update({'User-Agent': USER_AGENT}) retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504]) adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) return session # ========== 页面解析 ========== class CategoryParser: @staticmethod def parse(html, current_url): soup = BeautifulSoup(html, 'lxml') subcats, pages, next_url = [], [], None # 子分类 subcat_div = soup.find('div', id='mw-subcategories') if subcat_div: for a in subcat_div.find_all('a', href=True): if a['href'].startswith('/wiki/Category:'): subcats.append(urljoin(BASE_URL, a['href'])) # 普通页面 pages_div = soup.find('div', id='mw-pages') if pages_div: for a in pages_div.find_all('a', href=True): href = a['href'] if href.startswith('/wiki/') and not href.startswith('/wiki/Category:'): title = href[6:] # 去掉 '/wiki/' if is_article_page(title): pages.append(title) # 下一页 next_link = soup.find('a', string=lambda t: t and 'next page' in t.lower()) if not next_link: next_link = soup.find('a', class_='mw-nextpage') if next_link and next_link.get('href'): next_url = urljoin(BASE_URL, next_link['href']) return list(dict.fromkeys(subcats)), list(dict.fromkeys(pages)), next_url # ========== 状态管理 ========== @dataclass class CrawlState: visited: Set[str] = field(default_factory=set) all_titles: Set[str] = field(default_factory=set) cat_pages: Dict[str, List[str]] = field(default_factory=dict) queue: deque = field(default_factory=deque) depth: Dict[str, int] = field(default_factory=dict) def save(self, path='checkpoint.json'): with open(path, 'w') as f: json.dump({ 'visited': list(self.visited), 'all_titles': list(self.all_titles), 'cat_pages': self.cat_pages, 'queue': list(self.queue), 'depth': self.depth }, f) def load(self, path='checkpoint.json'): try: with open(path, 'r') as f: d = json.load(f) self.visited = set(d['visited']) self.all_titles = set(d['all_titles']) self.cat_pages = d['cat_pages'] self.queue = deque(d['queue']) self.depth = d['depth'] return True except FileNotFoundError: return False # ========== 主爬虫 ========== class WikiCrawler: def __init__(self, start_url, max_depth=MAX_DEPTH, delay=REQUEST_DELAY): self.start_url = start_url self.max_depth = max_depth self.delay = delay self.session = create_session() self.parser = CategoryParser() self.state = CrawlState() logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') self.log = logging.getLogger(__name__) def fetch(self, url): self.log.info(f"Fetching {url}") try: resp = self.session.get(url, timeout=30) resp.raise_for_status() return resp.text except Exception as e: self.log.error(f"Error fetching {url}: {e}") return None def crawl(self, resume=False): if resume and self.state.load(): self.log.info("Resuming from checkpoint") else: self.state.queue.append(self.start_url) self.state.depth[self.start_url] = 0 while self.state.queue: url = self.state.queue.popleft() depth = self.state.depth[url] if depth > self.max_depth: self.log.info(f"Max depth reached at {url}, skipping") continue if url in self.state.visited: continue html = self.fetch(url) if not html: continue subcats, pages, next_url = self.parser.parse(html, url) if pages: self.state.cat_pages[url] = pages for p in pages: self.state.all_titles.add(p) self.log.info(f"Found {len(pages)} pages in {url}") self.state.visited.add(url) for sc in subcats: if sc not in self.state.visited and sc not in self.state.depth: self.state.depth[sc] = depth + 1 self.state.queue.append(sc) if next_url and next_url not in self.state.visited and next_url not in self.state.depth: self.state.depth[next_url] = depth self.state.queue.append(next_url) time.sleep(self.delay) if len(self.state.visited) % 10 == 0: self.state.save() self.log.info(f"Done. Categories: {len(self.state.visited)}, Pages: {len(self.state.all_titles)}") return self.state.all_titles, self.state.cat_pages def save_results(self, txt_path='titles.txt', json_path='results.json'): with open(txt_path, 'w', encoding='utf-8') as f: for title in sorted(self.state.all_titles): f.write(title + '\n') with open(json_path, 'w', encoding='utf-8') as f: json.dump({ 'start': self.start_url, 'max_depth': self.max_depth, 'total_categories': len(self.state.visited), 'total_titles': len(self.state.all_titles), 'category_pages': self.state.cat_pages, 'all_titles': list(self.state.all_titles) }, f, indent=2, ensure_ascii=False) self.log.info(f"Saved to {txt_path} and {json_path}") # ========== 主程序入口 ========== if __name__ == '__main__': # 示例：爬取Machine Learning分类下所有页面（深度2层） start_category = "https://en.wikipedia.org/wiki/Category:Machine_learning" crawler = WikiCrawler(start_category, max_depth=2, delay=1.0) titles, mapping = crawler.crawl(resume=False) # 设为True可断点续爬 crawler.save_results() # 打印前20个标题作为示例 print("\n=== Sample Page Titles ===") for i, title in enumerate(sorted(titles)[:20]): print(f"{i+1}. {title}")

六、运行与测试

6.1 基本运行

将上述代码保存为wiki_crawler.py，在终端执行：

bash

python wiki_crawler.py

输出示例：

text

2025-01-15 10:23:11,456 - INFO - Fetching https://en.wikipedia.org/wiki/Category:Machine_learning 2025-01-15 10:23:12,789 - INFO - Found 45 pages in https://en.wikipedia.org/wiki/Category:Machine_learning 2025-01-15 10:23:13,801 - INFO - Fetching https://en.wikipedia.org/wiki/Category:Deep_learning ...

爬取完成后，生成titles.txt（所有页面标题，每行一个）和results.json（详细结果）。

6.2 测试不同分类

您可以修改start_category变量来爬取其他分类：

python

# 人工智能 start = "https://en.wikipedia.org/wiki/Category:Artificial_intelligence" # 中国历史 start = "https://en.wikipedia.org/wiki/Category:History_of_China" # 编程语言 start = "https://en.wikipedia.org/wiki/Category:Programming_languages"

6.3 调整深度

注意维基百科分类树非常庞大，max_depth=3时可能产生数千个子分类，数万个页面。建议从小深度开始测试：

python

crawler = WikiCrawler(start_category, max_depth=1) # 只爬直接子分类

七、性能优化与最佳实践

7.1 异步爬取

对于大规模爬取，同步请求会成为瓶颈。我们可以使用aiohttp+asyncio实现并发。核心改动：

python

import aiohttp import asyncio async def fetch_async(session, url): async with session.get(url) as resp: return await resp.text() async def crawl_async(self): async with aiohttp.ClientSession(headers=self.headers) as session: tasks = [self.process_category(session, url) for url in self.state.queue] await asyncio.gather(*tasks)

但异步会大大提高请求速度，必须相应调大请求间隔或使用维基百科的API（更友好）。

7.2 使用维基百科API代替HTML解析

维基百科提供了强大的REST API，可以JSON格式直接获取分类成员，效率远高于解析HTML。API端点：

text

https://en.wikipedia.org/w/api.php?action=query&list=categorymembers&cmtitle=Category:Machine_learning&cmtype=subcat|page&format=json

参数说明：

cmtitle：分类名称
cmtype=subcat|page：同时获取子分类和页面
cmlimit：每页数量（最大500）
cmcontinue：分页续传

使用API的示例代码片段：

python

import requests def get_category_members_api(category_name, cmcontinue=None): """通过API获取分类成员，category_name不含'Category:'前缀""" params = { 'action': 'query', 'list': 'categorymembers', 'cmtitle': f'Category:{category_name}', 'cmtype': 'subcat|page', 'cmlimit': 500, 'format': 'json' } if cmcontinue: params['cmcontinue'] = cmcontinue resp = requests.get('https://en.wikipedia.org/w/api.php', params=params) data = resp.json() pages = [] subcats = [] for member in data['query']['categorymembers']: if member['ns'] == 0: # 主命名空间 pages.append(member['title']) elif member['ns'] == 14: # 分类命名空间 subcats.append(member['title']) continue_token = data.get('continue', {}).get('cmcontinue') return pages, subcats, continue_token

API方式优点：