爬虫
爬虫要做的是:给定Luogu上的题目url,自动爬取题目和题解,并进行一定的解析。
省流:对于题目,用 Crawl4ai 渲染题目页,调用 LLM 按 Pydantic schema 提取并结构化题目信息;对于题解,先用 requests + BeautifulSoup 在
luogu.com.cn搜索文章拿到/article/...链接,再用 Crawl4ai 渲染这些公开文章并用 LLM 按 schema 提取题解与代码(绕过需登录的/solution/页面)。
工具介绍
首先,应该去GitHub上看有没有别人写好的Luogu爬虫,很可惜我直到开始写代码一个小时后才想起来去这么干,幸好GitHub上面没有能用的。
经过多次试错,这个项目最终演变成了一个“混合动力”的爬虫系统,没有一个工具是万能的,我用了三种工具,让它们各司其职:
Crawl4ai
介绍
它本质上是一个 Playwright(无头浏览器)+ LLM(大语言模型)的组合体。
- 优点: 不必像requests库那样写请求标头,并且可以让LLM来解析内容。内置了强大的方法将html转为markdown,使用起来非常方便。
- 缺点: 处理速度慢,且需要token消耗。因此在结构化网页最好不要使用LLM。
安装
# Install the package
pip install -U crawl4ai
# For pre release versions
pip install crawl4ai --pre
# Run post-installation setup
crawl4ai-setup
# Verify your installation
crawl4ai-doctor
requests
介绍
这是基本的HTTP请求库。 * 优点: 极快,轻量,0 Token消耗。 * 缺点: 它只能抓取静态的HTML源码,无法执行JavaScript。如果页面是动态渲染的,它就抓瞎了。
安装
pip install requests
BeautifulSoup
介绍
用于解析HTML源码。
* 优点: 能通过CSS选择器或正则表达式(re)精确地“抠”出想要的任何标签或链接。
* 缺点: 强依赖于固定的HTML结构。如果网站一改版,代码就得重写。
安装
pip install beautifulsoup4
题目页爬取
爬取题目页(.../problem/P1254)是第一步。这个页面是公开的,不需要登录。
虽然
requests+beautifulsoup也能抓取,但我一开始就选择了Crawl4ai。主要原因是,洛谷的题目格式并不统一(比如有的题没有“题目背景”,有的是中英文混杂)。另一个原因是之后或许可以改改用于其他网站。
首先,定义需要的字段:题目背景、题目描述、输入格式、输出格式、输入输出样例、说明/提示。
from pydantic import BaseModel, Field
from typing import List
class ProblemDetails(BaseModel):
background: str = Field(..., description="提取题目背景,如果没有则返回 '无'")
description: str = Field(..., description="提取完整的题目描述")
input_format: str = Field(..., description="提取输入格式说明")
output_format: str = Field(..., description="提取输出格式说明")
examples: List[dict] = Field(..., description="提取所有的输入输出样例。每个样例是一个字典,包含 'input' 和 'output' 键")
notes: str = Field(..., description="提取说明/提示部分。注意:必须保留所有的数学符号、上标和格式。如果没有则返回 '无'")
接着,定义LLM及其提取策略。我看了crawl4ai支持的模型,选了最便宜的deepseek。
from crawl4ai import *
from constant import DMX_API_KEY
strategy = LLMExtractionStrategy(
llm_config=LLMConfig(
provider="deepseek/deepseek-chat",
api_token=DMX_API_KEY,
base_url="https://www.dmxapi.cn/v1"
),
schema=ProblemDetails.model_json_schema(),
extraction_type="schema",
instruction="你是一个爬虫助手,请从给定的网页 Markdown 中提取算法题目的关键信息。", #
apply_chunking=False,
input_format="markdown", # 这里可能需要html, 目前先暂时这样
extra_args={"temperature": 0.1, "max_tokens": 2000}
)
在主函数内,异步地进行爬取,delay_before_return_html的设置是为了让js动态加载。
import json, asyncio, os
test_url = "https://www.luogu.com.cn/problem/P1254"
async def main():
brouser_config = BrowserConfig(headless=True, proxy=None)
config = CrawlerRunConfig(
cache_mode=CacheMode.ENABLED,
extraction_strategy=strategy,
delay_before_return_html=3
)
async with AsyncWebCrawler(config=brouser_config) as crawler:
print(f"正在爬取 {test_url} 并提取...")
result = await crawler.arun(test_url, config=config)
if result.success and result.extracted_content:
try:
data = json.loads(result.extracted_content)
print("--- 提取成功 ---")
print(json.dumps(data, indent=2, ensure_ascii=False))
# 打印 Token 消耗
strategy.show_usage()
except json.JSONDecodeError:
print("--- 提取失败 ---")
print("LLM 返回的不是一个有效的 JSON:")
print(result.extracted_content)
else:
print("--- 提取失败 ---")
print("爬虫或提取步骤出错。")
if result.error_message:
print(f"错误信息: {result.error_message}")
if __name__ == "__main__":
asyncio.run(main())
完整代码:
from pydantic import BaseModel, Field
import json, asyncio, os
from crawl4ai import *
from typing import List
from constant import DMX_API_KEY
class ProblemDetails(BaseModel):
background: str = Field(..., description="提取题目背景,如果没有则返回 '无'")
description: str = Field(..., description="提取完整的题目描述")
input_format: str = Field(..., description="提取输入格式说明")
output_format: str = Field(..., description="提取输出格式说明")
examples: List[dict] = Field(..., description="提取所有的输入输出样例。每个样例是一个字典,包含 'input' 和 'output' 键")
notes: str = Field(..., description="提取说明/提示部分。注意:必须保留所有的数学符号、上标和格式。如果没有则返回 '无'")
strategy = LLMExtractionStrategy(
llm_config=LLMConfig(
provider="deepseek/deepseek-chat",
api_token=DMX_API_KEY,
base_url="https://www.dmxapi.cn/v1"
),
schema=ProblemDetails.model_json_schema(),
extraction_type="schema",
instruction="你是一个爬虫助手,请从给定的网页 Markdown 中提取算法题目的关键信息。", #
apply_chunking=False,
input_format="markdown", # 这里可能需要html, 目前先暂时这样
extra_args={"temperature": 0.1, "max_tokens": 2000}
)
test_url = "https://www.luogu.com.cn/article/syub6c8r"
async def main():
brouser_config = BrowserConfig(headless=True, proxy=None)
config = CrawlerRunConfig(
cache_mode=CacheMode.ENABLED,
extraction_strategy=strategy,
delay_before_return_html=3
)
async with AsyncWebCrawler(config=brouser_config) as crawler:
print(f"正在爬取 {test_url} 并提取...")
result = await crawler.arun(test_url, config=config)
if result.success and result.extracted_content:
try:
data = json.loads(result.extracted_content)
print("--- 提取成功 ---")
print(json.dumps(data, indent=2, ensure_ascii=False))
# 打印 Token 消耗
strategy.show_usage()
except json.JSONDecodeError:
print("--- 提取失败 ---")
print("LLM 返回的不是一个有效的 JSON:")
print(result.extracted_content)
else:
print("--- 提取失败 ---")
print("爬虫或提取步骤出错。")
if result.error_message:
print(f"错误信息: {result.error_message}")
if __name__ == "__main__":
asyncio.run(main())
题解页爬取
对于题目的爬取方法在爬取题解的时候失效了,因为直接访问 .../solution/ 页面需要登陆。
这开启了我漫长的“踩坑”之旅。
方案一:直接爬取 (失败)
- 尝试: 保持和爬取题目页一样的逻辑,直接用
Crawl4ai访问.../solution/P1254。 - 结果:
Crawl4ai底层的浏览器因为没有登录,被洛谷的登录墙挡住,只爬回来“请先登录”的HTML。
方案二:注入 Cookie (失败)
- 尝试: 既然要登录,我就在浏览器上手动登录,然后按
F12抓到了我的登录 Cookie(_uid,__client_id等)[cite:_uid: 1022761,__client_id: 0b76e9e1441a36f9b7a73f0436accc308dfcd6fb,C3VK: 957d42]。然后,我使用Crawl4ai提供的on_page_context_created钩子,在浏览器导航前就把我的 Cookie 注入进去。 - 结果: 依然失败。具体原因尚不明确,因为hook的具体实现我没仔细看,也可能是实现有问题。gemini说是洛谷使用了“双重保险”:它不仅验证 Cookie,还验证一个动态的
CSRF-Token(你是否从洛谷网站本身发起的请求)。我爬虫虽然“登录”了,但被当成了CSRF攻击,还是被拒了。
最终方案:Requests + BeautifulSoup + Crawl4ai
我看了一会url之后,想起来Luogu有一个 luogu.com 和一个 luogu.com.cn 。我寻思这两个应该是有所不同的,luogu.com 我记得是2025年年初做的,好像是专门用来存放讨论区和文章,幸运的是,这个是不需要登陆的。唯一的问题在于我们需要在文章区通过搜索找到题目对应的题解url。
**阶段一:requests + BeautifulSoup **
用 requests 下载搜索页 (.../article?keyword=P1254),然后用 BeautifulSoup 配合正则表达式 re.compile(r'^/article/...'),精确地把所有文章链接(如 /article/syub6c8r) 提取出来。为了确保文章标题里包含"P1254" 再添加一个子串判断,排除搜索到的别的文章。
# --- 第 1 阶段:搜索 (requests + bs4) ---
async def search_for_solution_urls(problem_id: str) -> List[dict]:
search_url = f"https://www.luogu.com.cn/article?keyword={problem_id}&page=1"
headers = {
"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"
}
print(f"[Phase 1] 正在用 requests 搜索: {search_url} ...")
try:
response = requests.get(search_url, headers=headers)
soup = BeautifulSoup(response.text, 'html.parser')
article_links = soup.find_all('a', href=re.compile(r'^/article/[a-zA-Z0-9]{8}$'))
if not article_links:
print("[Phase 1] 失败:没有找到任何 /article/... 链接。")
return []
results = []
for link in article_links:
href = link.get('href')
title = link.get_text(strip=True)
if href and (problem_id in title or problem_id in href):
if href not in [r['url'] for r in results]:
results.append({"title": title, "url": href})
print(f"[Phase 1] 成功:找到 {len(results)} 篇相关题解链接。")
return results
except requests.RequestException as e:
print(f"[Phase 1] 异常: {e}")
return []
**阶段二:Crawl4ai **
到这里已经拿到阶段一产出的干净的文章 URL 列表。现在,Crawl4ai 终于可以上场了。我们循环这个列表,让 Crawl4ai 访问这些公开的文章页。
这次,我们使用了“宽容”的 SolutionDetails Schema,只要求 solution_text 必须存在,author 和 code 都是可选的,这大大提高了LLM的提取成功率。
最后,我们用 SolutionDetails.model_validate(raw_data) 来“清洗”LLM返回的、可能带有 {'error': false} 字段的JSON(目前原因尚不明确),得到可以交付的干净数据。
# --- 第 2 阶段:提取 (crawl4ai + LLM) ---
class SolutionDetails(BaseModel):
solution_text: str = Field(..., description="提取完整的题解思路和文字说明 (必须保留所有 LaTeX 数学公式)")
author: Optional[str] = Field(None, description="提取题解的作者用户名")
code: Optional[str] = Field(None, description="提取该题解对应的完整代码块")
code_language: Optional[str] = Field(None, description="提取代码块的编程语言 (例如 C++, Python)")
strategy = LLMExtractionStrategy(
llm_config=LLMConfig(
provider="deepseek/deepseek-chat",
api_token=DMX_API_KEY,
base_url="https://www.dmxapi.cn/v1"
),
schema=SolutionDetails.model_json_schema(),
extraction_type="schema",
instruction="你是一个爬虫助手,请从给定的文章页中提取作者、题解思路和代码。",
apply_chunking=False,
input_format="markdown",
extra_args={"temperature": 0.1, "max_tokens": 4096}
)
async def extract_solution_content(article_url: str, crawler: AsyncWebCrawler) -> dict:
"""
使用 crawl4ai 提取一篇公开文章页的具体内容。
"""
full_url = f"https://www.luogu.com.cn{article_url}"
print(f"[Phase 2] 正在用 crawl4ai 提取: {full_url} ...")
config = CrawlerRunConfig(
cache_mode=CacheMode.BYPASS,
extraction_strategy=strategy,
delay_before_return_html=3
)
result = await crawler.arun(full_url, config=config)
if result.success and result.extracted_content:
try:
data_list = json.loads(result.extracted_content)
if not data_list:
return {"error": "LLM 返回了空列表"}
raw_data = data_list[0]
clean_data_model = SolutionDetails.model_validate(raw_data)
return clean_data_model.model_dump()
except json.JSONDecodeError:
return {"error": "LLM 返回了无效的 JSON", "raw_content": result.extracted_content}
except Exception as e:
return {"error": f"Pydantic 验证失败: {e}", "raw_data": raw_data}
else:
return {"error": "crawl4ai 爬取失败"}
async def main():
PROBLEM_ID = "P1268"
article_urls = await search_for_solution_urls(PROBLEM_ID)
if not article_urls:
print("没有找到任何题解 URL,程序退出。")
return
print(f"\n--- 准备提取 {len(article_urls)} 篇文章的详细内容 ---")
brouser_config = BrowserConfig(headless=True, proxy=None)
async with AsyncWebCrawler(config=brouser_config) as crawler:
all_solutions = []
for article in article_urls[:3]:
solution_data = await extract_solution_content(article['url'], crawler)
if "error" in solution_data:
print(f"提取 {article['url']} 失败: {solution_data['error']}")
elif not solution_data.get("solution_text"):
print(f"提取 {article['url']} 失败: LLM 返回了空内容。")
else:
solution_data['title'] = article['title']
solution_data['url'] = article['url']
all_solutions.append(solution_data)
await asyncio.sleep(1) # 别爬太快
print("\n\n--- 爬取完毕:所有提取成功的题解 ---")
print(json.dumps(all_solutions, indent=2, ensure_ascii=False))
print(f"\n成功从 {len(article_urls)} 个链接中提取了 {len(all_solutions)} 篇题解。")
if __name__ == "__main__":
asyncio.run(main())
完整代码:
import requests
from bs4 import BeautifulSoup
import re
import json
import asyncio
import os
from pydantic import BaseModel, Field
from typing import List, Optional
from crawl4ai import *
from constant import DMX_API_KEY
# --- 第 1 阶段:搜索 (requests + bs4) ---
async def search_for_solution_urls(problem_id: str) -> List[dict]:
search_url = f"https://www.luogu.com.cn/article?keyword={problem_id}&page=1"
headers = {
"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"
}
print(f"[Phase 1] 正在用 requests 搜索: {search_url} ...")
try:
response = requests.get(search_url, headers=headers)
soup = BeautifulSoup(response.text, 'html.parser')
article_links = soup.find_all('a', href=re.compile(r'^/article/[a-zA-Z0-9]{8}$'))
if not article_links:
print("[Phase 1] 失败:没有找到任何 /article/... 链接。")
return []
results = []
for link in article_links:
href = link.get('href')
title = link.get_text(strip=True)
if href and (problem_id in title or problem_id in href):
if href not in [r['url'] for r in results]:
results.append({"title": title, "url": href})
print(f"[Phase 1] 成功:找到 {len(results)} 篇相关题解链接。")
return results
except requests.RequestException as e:
print(f"[Phase 1] 异常: {e}")
return []
# --- 第 2 阶段:提取 (crawl4ai + LLM) ---
class SolutionDetails(BaseModel):
solution_text: str = Field(..., description="提取完整的题解思路和文字说明 (必须保留所有 LaTeX 数学公式)")
author: Optional[str] = Field(None, description="提取题解的作者用户名")
code: Optional[str] = Field(None, description="提取该题解对应的完整代码块")
code_language: Optional[str] = Field(None, description="提取代码块的编程语言 (例如 C++, Python)")
strategy = LLMExtractionStrategy(
llm_config=LLMConfig(
provider="deepseek/deepseek-chat",
api_token=DMX_API_KEY,
base_url="https://www.dmxapi.cn/v1"
),
schema=SolutionDetails.model_json_schema(),
extraction_type="schema",
instruction="你是一个爬虫助手,请从给定的文章页中提取作者、题解思路和代码。",
apply_chunking=False,
input_format="markdown",
extra_args={"temperature": 0.1, "max_tokens": 4096}
)
async def extract_solution_content(article_url: str, crawler: AsyncWebCrawler) -> dict:
"""
使用 crawl4ai 提取一篇公开文章页的具体内容。
"""
full_url = f"https://www.luogu.com.cn{article_url}"
print(f"[Phase 2] 正在用 crawl4ai 提取: {full_url} ...")
config = CrawlerRunConfig(
cache_mode=CacheMode.BYPASS,
extraction_strategy=strategy,
delay_before_return_html=3
)
result = await crawler.arun(full_url, config=config)
if result.success and result.extracted_content:
try:
data_list = json.loads(result.extracted_content)
if not data_list:
return {"error": "LLM 返回了空列表"}
raw_data = data_list[0]
clean_data_model = SolutionDetails.model_validate(raw_data)
return clean_data_model.model_dump()
except json.JSONDecodeError:
return {"error": "LLM 返回了无效的 JSON", "raw_content": result.extracted_content}
except Exception as e:
return {"error": f"Pydantic 验证失败: {e}", "raw_data": raw_data}
else:
return {"error": "crawl4ai 爬取失败"}
async def main():
PROBLEM_ID = "P1268"
article_urls = await search_for_solution_urls(PROBLEM_ID)
if not article_urls:
print("没有找到任何题解 URL,程序退出。")
return
print(f"\n--- 准备提取 {len(article_urls)} 篇文章的详细内容 ---")
brouser_config = BrowserConfig(headless=True, proxy=None)
async with AsyncWebCrawler(config=brouser_config) as crawler:
all_solutions = []
for article in article_urls[:3]:
solution_data = await extract_solution_content(article['url'], crawler)
if "error" in solution_data:
print(f"提取 {article['url']} 失败: {solution_data['error']}")
elif not solution_data.get("solution_text"):
print(f"提取 {article['url']} 失败: LLM 返回了空内容。")
else:
solution_data['title'] = article['title']
solution_data['url'] = article['url']
all_solutions.append(solution_data)
await asyncio.sleep(1) # 别爬太快
print("\n\n--- 爬取完毕:所有提取成功的题解 ---")
print(json.dumps(all_solutions, indent=2, ensure_ascii=False))
print(f"\n成功从 {len(article_urls)} 个链接中提取了 {len(all_solutions)} 篇题解。")
if __name__ == "__main__":
asyncio.run(main())