跳转至

爬虫

爬虫要做的是:给定Luogu上的题目url,自动爬取题目和题解,并进行一定的解析。

省流:对于题目,用 Crawl4ai 渲染题目页,调用 LLM 按 Pydantic schema 提取并结构化题目信息;对于题解,先用 requests + BeautifulSoupluogu.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。
  • 尝试: 既然要登录,我就在浏览器上手动登录,然后按 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 必须存在,authorcode 都是可选的,这大大提高了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())