批量请求 - CFspider API 文档

概述

CFspider 提供强大的批量请求功能，支持并发控制、进度显示、失败重试和结果聚合，可以高效地处理大量 URL 请求。

cfspider.batch()

同步批量请求多个 URL，使用线程池实现并发。

函数签名

python

cfspider.batch(
    urls: Union[List[str], str],
    pick: Dict[str, Any] = None,
    concurrency: int = 5,
    delay: float = 0.0,
    retry: int = 0,
    timeout: float = 30.0,
    cf_proxies: str = None,
    token: str = None,
    impersonate: str = None,
    stealth: bool = False,
    stealth_browser: str = None,
    headers: Dict[str, str] = None,
    on_success: Callable = None,
    on_error: Callable = None,
    progress: bool = True,
    **kwargs
) -> BatchResult

参数说明

参数	类型	默认值	说明
`urls`	`List[str] / str`	必填	URL 列表或文件路径：列表：直接传入 URL 列表，如 `["url1", "url2"]` 字符串：文件路径，每行一个 URL，以 # 开头的行会被忽略
`pick`	`Dict[str, Any]`	`None`	数据提取规则字典，格式与 `response.pick()` 相同： `{"title": "h1"}`：提取标题 `{"links": ("a", "href")}`：提取链接属性 `{"price": (".price", "text", float)}`：提取并转换类型
`concurrency`	`int`	`5`	并发数，同时执行的请求数量。建议根据目标服务器承受能力设置，通常 5-20 之间
`delay`	`float`	`0.0`	请求间隔（秒），每个请求之间的最小时间间隔。用于避免请求过快被限制
`retry`	`int`	`0`	失败重试次数。请求失败时会自动重试，最多重试指定次数
`timeout`	`float`	`30.0`	单个请求的超时时间（秒）
`cf_proxies`	`str`	`None`	Workers 代理地址，所有请求都会通过此代理
`token`	`str`	`None`	Workers API 鉴权 token
`impersonate`	`str`	`None`	TLS 指纹模拟，如 "chrome131"
`stealth`	`bool`	`False`	是否启用隐身模式
`stealth_browser`	`str`	`None`	隐身模式浏览器类型
`headers`	`Dict[str, str]`	`None`	自定义请求头，所有请求都会使用这些请求头
`on_success`	`Callable`	`None`	成功回调函数，格式：`(url, response, data) -> None` `url`：请求的 URL `response`：响应对象 `data`：提取的数据（如果使用了 pick）
`on_error`	`Callable`	`None`	错误回调函数，格式：`(url, error) -> None` `url`：请求的 URL `error`：异常对象
`progress`	`bool`	`True`	是否显示进度条。需要安装 `tqdm`：`pip install tqdm`
`**kwargs`	`dict`	-	传递给 `cfspider.get()` 的其他参数

返回值

BatchResult

批量请求结果集合，支持迭代、过滤和导出。

使用示例

基本批量请求

python

import cfspider

urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3"
]

results = cfspider.batch(urls)
print(f"成功: {len(results.successful)}")
print(f"失败: {len(results.failed)}")

带数据提取的批量请求

python

results = cfspider.batch(
    urls,
    pick={
        "title": "h1",
        "links": ("a", "href"),
        "price": (".price", "text", float)
    },
    concurrency=10
)

# 保存结果
results.save("results.csv")

从文件读取 URL

python

# urls.txt 内容：
# https://example.com/page1
# https://example.com/page2
# # 这是注释，会被忽略
# https://example.com/page3

results = cfspider.batch("urls.txt", pick={"title": "h1"})

使用代理和重试

python

results = cfspider.batch(
    urls,
    cf_proxies="https://your-workers.dev",
    token="your-token",
    retry=3,  # 失败重试 3 次
    delay=1.0,  # 请求间隔 1 秒
    concurrency=5
)

使用回调函数

python

def on_success(url, response, data):
    print(f"✓ {url}: {data.get('title', 'N/A')}")

def on_error(url, error):
    print(f"✗ {url}: {error}")

results = cfspider.batch(
    urls,
    pick={"title": "h1"},
    on_success=on_success,
    on_error=on_error
)

cfspider.abatch()

异步批量请求多个 URL，使用 asyncio 实现并发，性能更高。

函数签名

python

await cfspider.abatch(
    urls: Union[List[str], str],
    pick: Dict[str, Any] = None,
    concurrency: int = 10,
    delay: float = 0.0,
    retry: int = 0,
    timeout: float = 30.0,
    cf_proxies: str = None,
    token: str = None,
    impersonate: str = None,
    stealth: bool = False,
    stealth_browser: str = None,
    headers: Dict[str, str] = None,
    on_success: Callable = None,
    on_error: Callable = None,
    progress: bool = True,
    **kwargs
) -> BatchResult

参数说明

参数与 batch() 相同，但 concurrency 默认值为 10（异步可以支持更高的并发数）。

使用示例

python

import asyncio
import cfspider

async def main():
    urls = ["https://example.com/page1", "https://example.com/page2"]
    
    results = await cfspider.abatch(
        urls,
        pick={"title": "h1"},
        concurrency=20  # 异步可以设置更高的并发数
    )
    
    results.save("results.csv")

asyncio.run(main())

BatchResult 对象

批量请求结果集合，提供丰富的方法和属性。

属性

属性	类型	说明
`successful`	`List[BatchItem]`	成功的结果列表
`failed`	`List[BatchItem]`	失败的结果列表
`success_rate`	`float`	成功率（0.0-1.0）

方法

方法	说明
`to_list()`	转换为字典列表，每个字典包含 url、提取的数据、错误信息等
`to_dataframe()`	转换为 pandas DataFrame（需要安装 pandas）
`save(filepath, **kwargs)`	保存结果到文件，支持 JSON/CSV/Excel/SQLite 格式
`filter(predicate)`	过滤结果，返回新的 BatchResult
`summary()`	获取结果摘要（总数、成功数、失败数、成功率、耗时等）

使用示例

python

results = cfspider.batch(urls, pick={"title": "h1"})

# 迭代结果
for item in results:
    if item.success:
        print(f"{item.url}: {item.data}")
    else:
        print(f"{item.url}: 失败 - {item.error}")

# 获取成功/失败的结果
print(f"成功: {len(results.successful)}")
print(f"失败: {len(results.failed)}")
print(f"成功率: {results.success_rate:.1%}")

# 过滤结果
successful_results = results.filter(lambda item: item.success)

# 获取摘要
summary = results.summary()
print(summary)
# {
#     'total': 100,
#     'successful': 95,
#     'failed': 5,
#     'success_rate': '95.0%',
#     'total_duration': '120.50s',
#     'avg_duration': '1.21s'
# }

# 保存结果
results.save("results.csv")
results.save("results.json")
results.save("results.xlsx")

# 转换为 DataFrame
df = results.to_dataframe()
print(df.head())

BatchItem 对象

单个请求的结果项。

属性

属性	类型	说明
`url`	`str`	请求的 URL
`data`	`Dict[str, Any]`	提取的数据（如果使用了 pick），否则为 None
`response`	`CFSpiderResponse`	响应对象（如果请求成功），否则为 None
`error`	`str`	错误信息（如果请求失败），否则为 None
`duration`	`float`	请求耗时（秒）
`success`	`bool`	请求是否成功（属性，非字段）

性能建议

并发数设置

同步 batch()：建议 5-20，受限于线程池
异步 abatch()：可以设置更高，建议 10-50，甚至更高
根据目标服务器的承受能力调整，避免被封禁

请求间隔

如果目标网站有反爬虫机制，建议设置 delay 参数，避免请求过快。通常 0.5-2 秒之间。

重试机制

网络请求可能因为各种原因失败，建议设置 retry=2 或更高，提高成功率。