How to build a more fast and stable crawler

#1 不要重复造轮子

RSSHub 是一个开源的 RSS 订阅服务，有很多开发者在上面开发了各种各样的订阅源，比如：

• twitter
• youtube
• 知乎
• 豆瓣
• 等等

这些都是高质量的数据源，如果自己从头开始写，不仅浪费时间，而且质量还可能不如这些现成的。我们只需要写程序去接收统一的 RSS 订阅源返回的数据，然后进行处理，就可以快速获取各个平台的数据。

拿Twitter举例，我们只需要提供 Twitter Cookie 配置到 RSSHub 中，访问 RSSHub 的 API http://localhost:1200/twitter/user/yihui_indie 就可以获取到 Twitter 用户 yihui_indie 的推文。

<item>
<title>Re @boo_hz 哈哈哈，其实是有的。但管他呢😬</title>
<description>Re @boo_hz 哈哈哈，其实是有的。但管他呢😬</description>
<link>https://x.com/yihui_indie/status/1870409487733149905</link>
<guid isPermaLink="false">https://twitter.com/yihui_indie/status/1870409487733149905</guid>
<pubDate>Sat, 21 Dec 2024 10:02:27 GMT</pubDate>
<author>熠辉 Indie</author>
</item>

<item>
<title>Re @hixiaoji 估计也是你收到产品建议最多的一条动态吧🤣</title>
<description>Re @hixiaoji 估计也是你收到产品建议最多的一条动态吧🤣</description>
<link>https://x.com/yihui_indie/status/1870335626970866059</link>
<guid isPermaLink="false">https://twitter.com/yihui_indie/status/1870335626970866059</guid>
<pubDate>Sat, 21 Dec 2024 05:08:57 GMT</pubDate>
<author>熠辉 Indie</author>
</item>

#2 借助LLM完成数据处理

我们获取一个平台的数据内容，通常需要对齐平台进行特制化处理，比如：

• 招聘平台需要提取出职位名称、职位链接、职位发布时间、职位公司、职位地点、职位薪资、职位要求等
• 知乎需要提取出回答内容、回答链接、回答发布时间、回答作者、回答点赞数、回答评论数等
• 等等

一般情况下，我们可以通过 browserless 很轻松可以获取网页的源数据(包裹HTML的数据内容)，然后通过 bs4 去除里面无用的标签，可以获得一份干净简短的HTML数据。这时候我们可以将这份数据交给LLM，让LLM从中提取出我们预期的 struct 数据。

举例：我想要获取 https://gmgn.ai/trade?chain=sol&tab=smart_degen 的聪明钱数据，我会利用 firecrawl 的 crawl api 完成这个任务

class GmgnTokenInfo(BaseModel):
    name: str
    volume: str
    holders: str
    volume_1h: str
    price: str


class GmgnTokenInfoList(BaseModel):
    tokens: list[GmgnTokenInfo]


def crawl_gmgn():
    app = firecrawl_app()
    data = app.scrape_url("https://gmgn.ai/?chain=sol", {
        'formats': ['extract'],
        'extract': {
            'schema': GmgnTokenInfoList.model_json_schema()
        }
    })
    print(json.dumps(data, indent=4))

输出

{
        "tokens": [
            {
                "name": "GOBLY",
                "volume": "$6.3K",
                "holders": "28,202",
                "volume_1h": "7.3K",
                "price": "$0.0\u20856389"
            },
            {
                "name": "USACOIN",
                "volume": "$11.4M",
                "holders": "21,530",
                "volume_1h": "472.1K",
                "price": "$0.0114"
            }
            ...
        ]
    }

这里我用到了 firecrawl 工具，这是一个很强大的爬虫工具，它可以获取任意网页的数据内容，这里我们使用它的 extract 功能，将 GmgnTokenInfoList 的结构丢给它，firecrwal 背后其实是调用 LLM 完成数据提取。

firecrawl scrape 详情可以参考：https://docs.firecrawl.dev/v1-welcome#extract-format

2024-12-28 更新

firecrawl 目前 extract 模式不太稳定，所以还是推荐自己写 agent 完成数据提取。 以下是我写的一个从 content 中提取出 struct 的 agent 示例：

import json
from typing import Any, Dict

import structlog

from src.util.openai_util import query_llm

logger = structlog.get_logger()


async def extract_structured_data(
    content: str, schema: Dict[str, Any]
) -> Dict[str, Any]:
    if not content.strip() or not schema:
        raise ValueError("Content and schema cannot be empty")

    # Construct the extraction prompt
    prompt = f"""
Please extract information from the following text and format it according to the schema.
Respond ONLY with a valid JSON object matching the schema structure.

Text content:
{content}

Required schema:
{json.dumps(schema, indent=2)}

Rules:
1. Extract all relevant information that matches the schema
2. Use null for missing values
3. Strictly follow the schema structure
4. Return only the JSON object, no additional text
"""

    try:
        # Query LLM with JSON response format
        response = await query_llm(query=prompt, response_json=True)

        # Parse and validate the response
        extracted_data = json.loads(response)

        return extracted_data

    except json.JSONDecodeError:
        raise Exception("Failed to parse LLM response as JSON")
    except Exception as e:
        raise Exception(f"Data extraction failed: {str(e)}")


if __name__ == "__main__":
    import asyncio

    # 示例文本和模式
    text_content = """
    John Doe is a software engineer at Tech Corp.
    He has 5 years of experience and specializes in Python and JavaScript.
    Contact: john.doe@email.com
    """

    schema = {
        "name": "string",
        "occupation": "string",
        "company": "string",
        "years_of_experience": "number",
        "skills": ["string"],
        "contact": {"email": "string"},
    }

    # 提取数据
    result = asyncio.run(extract_structured_data(text_content, schema))
    print(result)