funboost-spider-crawling

# Funboost 爬虫 / 数据采集 ## 概述 Funboost 是**函数调度器**，Scrapy 是 **URL 调度器**——这是二者最本质的差异。 - **Scrapy**：调度 `Request` 对象，框架替你发请求、走 callback 链、Pipeline 存储，用户被框架生命周期束缚。 - **Funboost**：调度**完整的 Python 函数**。框架只负责：何时调用、并发几个、失败怎么办。函数内部写什么代码完全自由。 **核心结论**：Funboost 写爬虫就是「写函数就能爬虫」。加一个 `@boost(BoosterParams(...))` 装饰器，函数自动获得分布式、并发、重试、控频、断点续爬、去重等 30+ 种能力。 **核心原则**：框架只管调度，不插手你怎么发请求、怎么解析、怎么存储。整个 PyPI 都是天然生态，无需 scrapy-xxx 插件。 ## 适用场景 - 分布式 Web 爬虫 / 数据采集 - 列表页 → 详情页 → 子任务链式派发 - 需要精准 QPS 控频、多进程/多机器横向扩展 - 短时效 Token、浏览器多轮交互（Playwright/Selenium） - 外部系统（Java/Go）通过 Redis/HTTP 实时触发爬取 - 函数级重试（HTTP 200 但内容异常也重试） ## 三种爬虫模式 | 模式 | 组成 | 适合场景 | |------|------|----------| | **纯 @boost** | `@boost` + 任意 HTTP 库（requests/httpx/playwright）+ 任意存储 | 最大自由，复用现有 `utils/` 工具 | | **funspider** | `SimpleSpiderClient` / `AsyncSpiderClient` + `SpiderResponse` + `SpiderItem` | ORM 模型驱动、强类型、同步/异步混用 | | **boost_spider** | `RequestClient` + `SpiderResponse` + `DatasetSink`（独立包 `pip install boost_spider`） | 极简字典流、多代理商容灾 | 三者**都不是必须的**，可自由混搭：`funboost + funspider(SpiderItem) + boost_spider(RequestClient) + your_utils`。 ### 模式 1：纯 @boost 函数调度 ```python from funboost import boost, BoosterParams, BrokerEnum @boost(BoosterParams( queue_name='crawl_detail', broker_kind=BrokerEnum.REDIS_ACK_ABLE, qps=5, max_retry_times=3, )) def crawl_detail(url: str): import requests resp = requests.get(url, proxies=get_my_proxy(), headers=get_my_headers()) data = parse_detail(resp.text) save_to_mysql(data) crawl_detail.push(url='https://example.com/article/123') crawl_detail.consume() ``` ### 模式 2：funspider（内置于 funboost） 位于 `funboost/contrib/funspider/`，依赖：`pip install httpx parsel sqlmodel`（及数据库驱动）。 | 组件 | 说明 | |------|------| | `SimpleSpiderClient` | 同步 httpx 客户端：自动重试、随机 UA、代理轮换 | | `AsyncSpiderClient` | 异步 httpx 客户端：适配 `ConcurrentModeEnum.ASYNC` | | `SpiderResponse` | 响应封装：`.xpath()` / `.css()` / `.re()` / `.re_first()` / `.resp_dict` | | `SpiderItem` | SQLModel ORM 基类：`upsert()` / `aio_upsert()` / `mongo_upsert()` / `bulk_upsert()` | ### 模式 3：boost_spider（独立三方包） ```bash pip install boost_spider ``` | 组件 | 说明 | |------|------| | `RequestClient` | API 兼容 requests，多代理商容灾轮换、自动重试、Cookie 会话 | | `SpiderResponse` | `.xpath()` / `.css()` / `.re_search()` | | `DatasetSink` | 一行代码存 MySQL/PostgreSQL/SQLite/MongoDB，自动建表 | **funspider vs boost_spider 选型**： | 维度 | funspider | boost_spider | |------|-----------|--------------| | 设计理念 | ORM 辅助、强类型流 | 自由至上、极简字典流 | | 数据模型 | SQLModel 类，IDE 自动补全 | 纯字典 + `DatasetSink` | | HTTP 客户端 | httpx（同步+异步双引擎） | 兼容 requests，多代理商容灾 | | 适合场景 | 大型团队、ORM 管理表结构 | 快速开发、极致简洁 | ## funspider 客户端用法 ### SimpleSpiderClient（同步） ```python from funboost.contrib.funspider import SimpleSpiderClient def abuyun_proxy(): return "http://user:pass@proxy.abuyun.com:9020" def redis_pool_proxy(): import redis r = redis.Redis(host="localhost", port=6379, db=0) proxy = r.srandmember("proxy_pool") return proxy.decode() if proxy else None client = SimpleSpiderClient( proxy_getter_list=[abuyun_proxy, redis_pool_proxy], # 轮换代理 retry_times=3, # 失败自动重试次数（默认 2） timeout=30, # 超时秒数 user_agents=None, # 默认内置 UA 列表，可自定义 ) resp = client.get("https://example.com/page") resp = client.post("https://api.example.com", json={"key": "val"}) resp = client.request("GET", url, headers={...}, params={...}) client.close() # 程序结束时关闭连接 ``` **SpiderResponse 解析**： ```python title = resp.css("h1::text").get() links = resp.css("a.title::attr(href)").getall() news_id = resp.re_first(r'/news/(\d+)') items = resp.xpath("//div[@class='item']//text()").getall() data = resp.resp_dict # JSON 响应自动解析为 dict ``` ### AsyncSpiderClient（异步） 不绑定 Event Loop，可在 funboost `ASYNC` 并发模式自由使用： ```python from funboost import ConcurrentModeEnum from funboost.contrib.funspider import AsyncSpiderClient async_client = AsyncSpiderClient(proxy_getter_list=[abuyun_proxy], retry_times=3) @boost(BoosterParams( queue_name='async_crawl', concurrent_mode=ConcurrentModeEnum.ASYNC, qps=10, )) async def crawl_api(news_id: int): resp = await async_client.get(f"https://api.example.com/comments/{news_id}") for c in resp.resp_dict["comments"]: await CommentItem(...).aio_upsert() # 程序结束时：await async_client.aclose() ``` **注意**：异步任务必须用 `await func.aio_push(...)` 发布，消费函数必须 `async def`。 ## SpiderItem 数据模型 继承 `SpiderItem` + `table=True`，绑定数据库引擎和 upsert 去重字段： ```python from typing import ClassVar, Optional from funboost.contrib.funspider import SpiderItem, Field, create_engine, create_async_engine MYSQL_ENGINE = create_engine("mysql+pymysql://root:pass@127.0.0.1:3306/spider_db") ASYNC_MYSQL_ENGINE = create_async_engine("mysql+aiomysql://root:pass@127.0.0.1:3306/spider_db") class NewsItem(SpiderItem, table=True): __tablename__: ClassVar[str] = "news" __engine__ = MYSQL_ENGINE # 同步引擎 __async_engine__ = ASYNC_MYSQL_ENGINE # 异步引擎（用 aio_upsert 时需要） __default_upsert_unique_fields__ = ["news_id"] # upsert 去重键 id: Optional[int] = Field(default=None, primary_key=True) news_id: int = Field(unique=True) title: str = Field(max_length=200) url: str = Field(max_length=500) content: str NewsItem.create_table() # 自动建表，已存在则跳过 ``` **入库方法**： | 方法 | 说明 | |------|------| | `.insert()` | 同步插入 | | `.upsert()` | 同步 upsert（按 `__default_upsert_unique_fields__` 去重，存在则更新） | | `.bulk_upsert(items)` | 同步批量 upsert | | `.aio_insert()` / `.aio_upsert()` | 异步版本 | | `.mongo_upsert()` / `.aio_mongo_upsert()` | MongoDB upsert（需设置 `__mongo_collection__`） | | `.ensure_mongo_indexes()` | 按去重字段创建 MongoDB 索引 | ```python NewsItem(news_id=123, title="标题", url="https://...", content="正文").upsert() await CommentItem(...).aio_upsert() ``` ## 分布式爬虫架构 典型架构：**多队列 + Redis ACK 队列 + 多进程/多机器消费**。 ``` 种子推送 → [news_list 队列] → crawl_list 函数 → push 详情 URL ↓ [news_detail 队列] → crawl_detail 函数 → upsert 入库 + push 子任务 ↓ [news_comments 队列] → crawl_comments 函数（异步） ``` **关键配置**： ```python from funboost import boost, BoosterParams, BrokerEnum, BoostersManager CRAWLER_GROUP = "news_crawler" class NewsCrawlerParams(BoosterParams): broker_kind: str = BrokerEnum.REDIS_ACK_ABLE # ACK 确认，崩溃不丢任务 booster_group: str = CRAWLER_GROUP # 消费分组，一键启动 @boost(NewsCrawlerParams(queue_name='news_list', qps=2, do_task_filtering=True)) def crawl_list(page: int): ... @boost(NewsCrawlerParams(queue_name='news_detail', qps=5, max_retry_times=5)) def crawl_detail(detail_url: str): ... if __name__ == '__main__': BoostersManager.consume_group(CRAWLER_GROUP) # 一键启动全组消费 crawl_list.push(page=1) # 极限性能：crawl_detail.multi_process_consume(8) # 8 进程 × concurrent_num 线程 ``` **横向扩展**： - **多进程**：`func.multi_process_consume(8)` — 单机 8 核 × 50 线程 = 400 并发 - **多机器**：多台机器运行同一脚本、同一 `queue_name`、同一 Redis，自动竞争消费 - **分布式控频**：`is_using_distributed_frequency_control=True` — 多机器共享 QPS 上限 - **外部触发**：Java/Go 往 Redis 对应 queue 发 JSON（函数入参）即可；或用 `funboost.faas` 暴露 HTTP 接口 **Broker 推荐**：生产爬虫用 `BrokerEnum.REDIS_ACK_ABLE`（消费确认，断点续爬）；开发测试可用 `SQLITE_QUEUE`。 ## 去重（do_task_filtering） 基于**函数入参**自动去重（非 URL 字符串指纹），可忽略 URL 中的噪音参数： ```python @boost(BoosterParams( queue_name='news_detail', do_task_filtering=True, # 开启入参级去重 task_filtering_expire_seconds=3600 * 24, # 去重有效期 24 小时（过期后可再次爬取） )) def crawl_detail(detail_url: str): ... ``` - 相同 `detail_url` 在有效期内不会重复执行 - 列表页和详情页可设不同去重有效期（列表 24h，详情 7 天） - 需要 Redis 支持（框架自动使用 Redis 存储去重键） - 与 Scrapy URL 指纹去重相比：基于结构化入参，不受 URL 参数顺序/噪音干扰 **链式派发 + 去重**：列表页 `crawl_detail.push(detail_url=url)` 时，若详情页已开启 `do_task_filtering`，重复 URL 自动跳过。 ## 完整代码示例（funspider） 参考源码：`funboost/contrib/funspider/funspider_demos/funspider_demo1.py` ```python import re from typing import ClassVar, Optional from funboost import ( boost, BoosterParams, BoostersManager, BrokerEnum, ConcurrentModeEnum, ) from funboost.contrib.funspider import ( SimpleSpiderClient, AsyncSpiderClient, SpiderItem, create_engine, create_async_engine, Field, ) NEWS_GROUP = "news_crawler" class NewsCrawlerParams(BoosterParams): broker_kind: str = BrokerEnum.REDIS_ACK_ABLE booster_group: str = NEWS_GROUP MYSQL_ENGINE = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/testdb") ASYNC_MYSQL_ENGINE = create_async_engine("mysql+aiomysql://root:123456@127.0.0.1:3306/testdb") class NewsItem(SpiderItem, table=True): __tablename__: ClassVar[str] = "news" __engine__ = MYSQL_ENGINE __async_engine__ = ASYNC_MYSQL_ENGINE __default_upsert_unique_fields__ = ["news_id"] id: Optional[int] = Field(default=None, primary_key=True) news_id: int = Field(unique=True) title: str content: str url: str class CommentItem(SpiderItem, table=True): __tablename__: ClassVar[str] = "comments" __engine__ = MYSQL_ENGINE __async_engine__ = ASYNC_MYSQL_ENGINE __default_upsert_unique_fields__ = ["comment_id"] id: Optional[int] = Field(default=None, primary_key=True) comment_id: int = Field(unique=True) news_id: int user: str content: str NewsItem.create_table() CommentItem.create_table() sync_client = SimpleSpiderClient(retry_times=3) async_client = AsyncSpiderClient(retry_times=3) BASE_URL = "https://news.example.com" @boost(NewsCrawlerParams(queue_name="news_list", qps=2)) def crawl_list(page: int): resp = sync_client.get(f"{BASE_URL}/list?page={page}") for href in resp.css("a.detail::attr(href)").getall(): crawl_detail.push(detail_url=f"{BASE_URL}{href}") if resp.css("a.next-page::attr(href)").get(): crawl_list.push(page=page + 1) @boost(NewsCrawlerParams(