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

一、项目背景与目标在数据科学和自然语言处理领域维基百科是一个极为宝贵的数据源。它包含了数以百万计的结构化知识条目覆盖了几乎所有学科领域。在实际应用中我们经常需要获取某个主题分类下的所有页面标题例如“机器学习”分类下的所有算法页面“人工智能”分类下的所有相关概念或者“中国城市”分类下的所有城市条目。这些页面标题可以用于构建知识图谱、训练词向量模型、进行主题建模或者作为更大规模爬虫的种子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 lxmllxml作为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_learning3.2 页面内容组织打开一个分类页面例如Machine learning分类页面主要包含三个区域页面说明区页面顶部通常包含分类的简短描述子分类区Subcategories包含该分类下的所有子分类每个子分类是一个链接格式为/wiki/Category:子分类名页面区Pages包含直接属于该分类的普通文章页面每个页面是一个链接格式为/wiki/页面标题此外大型分类会分页显示页面底部会有“下一页”链接next page。3.3 HTML标签定位通过浏览器开发者工具F12分析我们可以发现子分类列表位于div idmw-subcategories中的div classCategoryTreeItem或直接是li标签下的链接页面列表位于div idmw-pages中的li标签下的链接下一页链接a href...next page/a位于div idmw-pages后的导航区域但在不同维基百科语言版本或不同主题下结构可能略有差异。为了健壮性我们会使用更通用的选择器。四、核心功能实现4.1 请求头与会话管理维基百科要求爬虫设置User-Agent标识自己否则可能返回403错误。同时为了效率我们复用TCP连接使用requests.Session。pythonimport 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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: en-US,en;q0.5, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, }) # 配置重试策略网络波动时自动重试 retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session4.2 页面下载与解析定义分类页面的解析器提取三类信息子分类链接、普通页面标题、下一页URL。pythonfrom 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) # 方法找到 idmw-subcategories 的div然后提取其中的链接 subcat_div soup.find(div, idmw-subcategories) if subcat_div: # 子分类可能在多种标签中常见的是 li 或 div.CategoryTreeItem for link in subcat_div.find_all(a, hrefTrue): href link[href] if href.startswith(/wiki/Category:): full_url urljoin(CategoryPageParser.BASE_URL, href) subcategories.append(full_url) # 备选方案如果上面没有找到尝试通过 classCategoryTreeItem 查找 if not subcategories: for item in soup.find_all(div, class_CategoryTreeItem): link item.find(a, hrefTrue) 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, idmw-pages) if pages_div: for link in pages_div.find_all(a, hrefTrue): 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... classmw-nextpagenext page/a next_link soup.find(a, stringlambda 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_url4.3 递归爬取与去重控制递归爬取的核心挑战循环引用分类A包含分类B分类B也包含分类A少见但存在重复爬取同一个分类可能被多个父分类引用深度爆炸维基百科分类体系非常庞大需要设置最大深度我们采用广度优先搜索BFS的策略配合visited_categories集合记录已处理的分类URL。pythonfrom 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_factoryset) all_page_titles: Set[str] field(default_factoryset) category_to_pages: Dict[str, List[str]] field(default_factorydict) queue: deque field(default_factorydeque) def save_checkpoint(self, filenamecrawler_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, encodingutf-8) as f: json.dump(data, f, indent2, ensure_asciiFalse) def load_checkpoint(self, filenamecrawler_checkpoint.json): 加载之前保存的进度 try: with open(filename, r, encodingutf-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 False4.4 主爬虫逻辑pythonclass WikipediaCategoryCrawler: 维基百科分类爬虫主类 def __init__(self, start_category_url, max_depth3, request_delay1.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( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) self.logger logging.getLogger(__name__) def fetch_page(self, url): 下载页面返回HTML文本 try: self.logger.info(fFetching: {url}) response self.session.get(url, timeout30) response.raise_for_status() # 维基百科返回的内容编码通常是utf-8 response.encoding utf-8 return response.text except requests.RequestException as e: self.logger.error(fFailed to fetch {url}: {e}) return None def crawl(self, resumeFalse): 开始爬取 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(fReached 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(fFound {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(fAdded subcategory: {subcat_url} (depth {current_depth1})) # 处理分页当前分类可能有多页下一页中的内容仍属于同一深度 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(fAdded 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(fCrawl completed. Visited {len(self.state.visited_categories)} categories, ffound {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_filepage_titles.txt, json_filecrawler_results.json): 保存爬取结果 # 保存为纯文本每行一个标题 with open(titles_file, w, encodingutf-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, encodingutf-8) as f: json.dump(results, f, indent2, ensure_asciiFalse) self.logger.info(fResults 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前缀来过滤pythondef 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.pypython#!/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(total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504]) adapter HTTPAdapter(max_retriesretry) 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, idmw-subcategories) if subcat_div: for a in subcat_div.find_all(a, hrefTrue): if a[href].startswith(/wiki/Category:): subcats.append(urljoin(BASE_URL, a[href])) # 普通页面 pages_div soup.find(div, idmw-pages) if pages_div: for a in pages_div.find_all(a, hrefTrue): 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, stringlambda 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_factoryset) all_titles: Set[str] field(default_factoryset) cat_pages: Dict[str, List[str]] field(default_factorydict) queue: deque field(default_factorydeque) depth: Dict[str, int] field(default_factorydict) def save(self, pathcheckpoint.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, pathcheckpoint.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_depthMAX_DEPTH, delayREQUEST_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(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) self.log logging.getLogger(__name__) def fetch(self, url): self.log.info(fFetching {url}) try: resp self.session.get(url, timeout30) resp.raise_for_status() return resp.text except Exception as e: self.log.error(fError fetching {url}: {e}) return None def crawl(self, resumeFalse): 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(fMax 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(fFound {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(fDone. 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_pathtitles.txt, json_pathresults.json): with open(txt_path, w, encodingutf-8) as f: for title in sorted(self.state.all_titles): f.write(title \n) with open(json_path, w, encodingutf-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, indent2, ensure_asciiFalse) self.log.info(fSaved 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_depth2, delay1.0) titles, mapping crawler.crawl(resumeFalse) # 设为True可断点续爬 crawler.save_results() # 打印前20个标题作为示例 print(\n Sample Page Titles ) for i, title in enumerate(sorted(titles)[:20]): print(f{i1}. {title})六、运行与测试6.1 基本运行将上述代码保存为wiki_crawler.py在终端执行bashpython wiki_crawler.py输出示例text2025-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_languages6.3 调整深度注意维基百科分类树非常庞大max_depth3时可能产生数千个子分类数万个页面。建议从小深度开始测试pythoncrawler WikiCrawler(start_category, max_depth1) # 只爬直接子分类七、性能优化与最佳实践7.1 异步爬取对于大规模爬取同步请求会成为瓶颈。我们可以使用aiohttpasyncio实现并发。核心改动pythonimport 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(headersself.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端点texthttps://en.wikipedia.org/w/api.php?actionquerylistcategorymemberscmtitleCategory:Machine_learningcmtypesubcat|pageformatjson参数说明cmtitle分类名称cmtypesubcat|page同时获取子分类和页面cmlimit每页数量最大500cmcontinue分页续传使用API的示例代码片段pythonimport requests def get_category_members_api(category_name, cmcontinueNone): 通过API获取分类成员category_name不含Category:前缀 params { action: query, list: categorymembers, cmtitle: fCategory:{category_name}, cmtype: subcat|page, cmlimit: 500, format: json } if cmcontinue: params[cmcontinue] cmcontinue resp requests.get(https://en.wikipedia.org/w/api.php, paramsparams) 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_tokenAPI方式优点无需解析HTML更稳定返回结构化JSON处理简单可以获取更多元数据页面ID、时间戳等请求更轻量不返回完整HTML7.3 遵守robots.txt和速率限制维基百科的robots.txt (https://en.wikipedia.org/robots.txt) 允许爬虫访问大部分内容但建议每秒不超过1个请求使用Accept-Encoding: gzip减少带宽设置From或Contact邮件头7.4 数据存储优化当页面标题数量达到数十万级别时内存中的set和dict可能消耗过大。可改用SQLite数据库存储Redis分布式场景分批写入磁盘八、常见问题与解决方案Q1: 遇到HTTP 429 Too Many Requests原因请求过快被限流。解决增大request_delay到2秒或更高并使用重试机制。Q2: 某些分类页面返回空内容或跳转原因分类可能被重定向或删除。解决检查响应URL是否改变使用response.history跟踪重定向。Q3: 递归深度过大导致内存爆炸原因维基百科分类树深度可能超过10层。解决设置合理的max_depth或改用迭代加深搜索。Q4: 特殊字符编码问题原因页面标题包含非ASCII字符如中文、日文。解决确保使用UTF-8编码requests默认处理良好保存文件时指定encodingutf-8。九、扩展应用爬取到的页面标题可以用于构建领域词典例如机器学习领域的所有术语列表。批量下载页面内容使用标题构造URL (https://en.wikipedia.org/wiki/标题)进一步爬取正文。生成站点地图为维基百科的某个主题区域建立索引。知识图谱实体抽取标题本身就是实体名称。比较不同语言版本爬取中文维基对应分类做跨语言对齐。十、总结本文详细实现了一个健壮的维基百科分类爬虫涵盖了以下关键技术点递归爬取策略使用BFS队列管理支持深度限制。去重与防循环通过visited集合和depth字典记录访问状态。分页处理识别下一页链接合并同分类下的多页内容。断点续爬定期保存进度到JSON支持中断后恢复。健壮性设计重试机制、异常捕获、日志记录。