爬虫-Crawlee

发表于 分类于 本文字数： 140k 阅读时长 ≈ 4:14

这篇讲什么

  • Crawlee for Python 1.11.1 在 Python 3.14.10 上逐条实测：HTTP 类爬虫（Basic/Http/Parsel/BeautifulSoup/FileDownload）、浏览器爬虫（Playwright/AdaptivePlaywright）、RequestQueue 与 Dataset 的本地存储、enqueue_links 四种策略、重试与错误处理、并发伸缩、代理与会话与指纹、robots.txt、停止与续爬。抓取对象沿用系列惯例：ssr1.scrape.center（服务端渲染）与 spa1.scrape.center（前端渲染），协议级用本机靶站。
  • 它是异步优先的框架：所有 handler 都是 async def，本地存储是 storage/ 目录下的 JSON 文件，没有 Scrapy 的 Pipeline/中间件那套，也没有 feapder 的任务表与批次表。
  • 装的时候 extras 决定你能用哪些类（裸装只有 HttpCrawler，报错原文见下一节），跑起来还有三个只有实测才知道的坑：默认队列会记住处理过的 URL，同一个 URL 第二次跑什么都不抓；enqueue_links(strategy=...) 只认四个小写值，写错不校验、把请求重试到失败；ConcurrencySettings(max_concurrency=1) 单独写会 ValueError，因为 desired_concurrency 默认是 10。

装哪种 extras

官方 quick-start 给的安装命令是一句 python -m pip install 'crawlee[all]'，用到浏览器爬虫再补一句 playwright install。文档页还列了按需要的装法：

1
2
3
4
5
6
7
8
python -m pip install crawlee
python -m pip install 'crawlee[all]'
python -m pip install 'crawlee[beautifulsoup]'
python -m pip install 'crawlee[parsel]'
python -m pip install 'crawlee[curl-impersonate]'
python -m pip install 'crawlee[playwright]'
python -m pip install 'crawlee[beautifulsoup,curl-impersonate]'
playwright install

先记两个数：crawlee[all] 在这台机器上要装 127 个包（scipy、scikit-learn、grpcio、stagehand、pydantic-ai 全在里面），裸装 crawlee 只有 21 个包。所以别一上来就 [all]，先看裸装缺什么——下面这段在一个只装了 crawlee 的干净虚拟环境里跑：

bare_install_surface.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# -*- coding: utf-8 -*-
"""裸装 crawlee（不带 extras）能用哪些类：报错原文照抄，它会把该装的 extra 名字告诉你。"""

import crawlee
import crawlee.crawlers as C

print("crawlee", crawlee.__version__)

checks = [
    ("BasicCrawler", lambda: C.BasicCrawler),
    ("HttpCrawler", lambda: C.HttpCrawler),
    ("FileDownloadCrawler", lambda: C.FileDownloadCrawler),
    ("ParselCrawler", lambda: C.ParselCrawler),
    ("BeautifulSoupCrawler", lambda: C.BeautifulSoupCrawler),
    ("PlaywrightCrawler", lambda: C.PlaywrightCrawler),
    ("AdaptivePlaywrightCrawler", lambda: C.AdaptivePlaywrightCrawler),
    ("HttpxHttpClient", lambda: __import__("crawlee.http_clients", fromlist=["x"]).HttpxHttpClient),
    ("CurlImpersonateHttpClient",
     lambda: __import__("crawlee.http_clients", fromlist=["x"]).CurlImpersonateHttpClient),
    ("DefaultFingerprintGenerator",
     lambda: __import__("crawlee.fingerprint_suite", fromlist=["x"]).DefaultFingerprintGenerator),
]

for label, getter in checks:
    try:
        getter()
        print("%-26s 可用" % label)
    except Exception as exc:
        print("%-26s %s: %s" % (label, type(exc).__name__, exc))
1
2
3
4
5
6
7
8
9
10
11
crawlee 1.11.1
BasicCrawler               可用
HttpCrawler                可用
FileDownloadCrawler        可用
ParselCrawler              ImportError: No module named 'parsel'. Install the optional 'parsel' extra to use it: pip install 'crawlee[parsel]'
BeautifulSoupCrawler       ImportError: No module named 'bs4'. Install the optional 'beautifulsoup' extra to use it: pip install 'crawlee[beautifulsoup]'
PlaywrightCrawler          ImportError: No module named 'playwright'. Install the optional 'playwright' extra to use it: pip install 'crawlee[playwright]'
AdaptivePlaywrightCrawler  ImportError: No module named 'playwright'. Install the optional 'adaptive-crawler' extra to use it: pip install 'crawlee[adaptive-crawler]'
HttpxHttpClient            ImportError: No module named 'httpx2'. Install the optional 'httpx' extra to use it: pip install 'crawlee[httpx]'
CurlImpersonateHttpClient  ImportError: No module named 'curl_cffi'. Install the optional 'curl-impersonate' extra to use it: pip install 'crawlee[curl-impersonate]'
DefaultFingerprintGenerator ModuleNotFoundError: No module named 'apify_fingerprint_datapoints'
  • 报错信息是设计过的：Install the optional 'parsel' extra to use it: pip install 'crawlee[parsel]'，照着补装即可；这类导入是延迟到取属性时才炸，所以 import crawlee.crawlers 本身不会失败。
  • 唯一一句没告诉你怎么办的是 DefaultFingerprintGenerator：它抛 ModuleNotFoundError: No module named 'apify_fingerprint_datapoints'，而这个包不在 httpx/playwright 之外的任何单装路径里——要指纹生成就装 crawlee[adaptive-crawler] 或 crawlee[all]。
  • pip install 'crawlee[sql_sqlite]' 与 'crawlee[sql-sqlite]' 两种写法实测都能装（文档里写的是下划线那版，PyPI 元数据里是连字符），pip 会做名字规范化；但 uv/poetry 未必兜得住，抄文档时留意。

本机这套实测环境的版本清单（importlib.metadata 现场读）：

env_versions.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import sys
from importlib.metadata import version

print("python :", sys.version.split()[0])
for pkg in ("crawlee", "impit", "pydantic", "httpx", "curl_cffi", "parsel",
            "beautifulsoup4", "playwright", "browserforge", "protego", "SQLAlchemy",
            "redis", "typer"):
    try:
        print("%-14s: %s" % (pkg, version(pkg)))
    except Exception:
        print("%-14s: 未安装" % pkg)

import crawlee
print("crawlee.__version__ :", crawlee.__version__)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
python : 3.14.10
crawlee       : 1.11.1
impit         : 0.15.1
pydantic      : 2.14.3
httpx         : 0.28.1
curl_cffi     : 0.17.0
parsel        : 1.12.1
beautifulsoup4: 4.15.0
playwright    : 1.64.1
browserforge  : 1.2.4
protego       : 0.7.0
SQLAlchemy    : 2.1.8
redis         : 8.1.0
typer         : 0.28.0
crawlee.__version__ : 1.11.1

第一个爬虫

官方最小示例的骨架就三段：建爬虫、注册默认 handler、crawler.run(种子列表)。这里把它落到本系列的靶站上，抓 ssr1.scrape.center 第一页：

first_crawler.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# -*- coding: utf-8 -*-
"""最小可用爬虫：官方 quick-start 的写法，抓 ssr1 第一页并把数据推进数据集。"""

import asyncio

from crawlee.crawlers import BeautifulSoupCrawler, BeautifulSoupCrawlingContext
from crawlee.storages import Dataset


async def main() -> None:
    crawler = BeautifulSoupCrawler(max_requests_per_crawl=1, configure_logging=False)

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        cards = context.soup.select("div.el-card")
        await context.push_data([{
            "title": card.select_one("h2").get_text().strip(),
            "score": card.select_one("p.score").get_text().strip(),
        } for card in cards])
        print("卡片数:", len(cards))

    await crawler.run(["https://ssr1.scrape.center/page/1"])

    items = await (await Dataset.open()).list_items()
    print("数据集条数:", len(items))
    print("第一条:", items[0])


asyncio.run(main())
1
2
3
卡片数: 10
数据集条数: 10
第一条: {'title': '霸王别姬 - Farewell My Concubine', 'score': '9.5'}
  • 每个 handler 都是 async def，crawler.run() 也是协程，所以最外面要 asyncio.run(main())；整篇没有一处同步写法可抄。
  • context.push_data(列表) 会把列表里的每一项当一条记录推进数据集，跑完在 storage/datasets/default/ 里能看到一条条 JSON。
  • max_requests_per_crawl=1 是「这次最多抓几个请求」的保险丝，调试时先设小，别一上来就顺着链接抓全站。

把同一个例子放大成 10 页 100 条，顺便看数据集读写与导出：

full_movie_crawl.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
# -*- coding: utf-8 -*-
"""完整例子：显式下发 10 个分页 URL，抓 100 部电影；看 context.push_data 落到哪个数据集，并导出 CSV。"""

import asyncio
import os
import shutil

from crawlee.crawlers import ParselCrawler
from crawlee.storages import Dataset, KeyValueStore, RequestQueue

BASE = "https://ssr1.scrape.center"
LOG = []
SEEDS = [f"{BASE}/page/{n}" for n in range(1, 11)]
named = None


async def main() -> None:
    global named
    shutil.rmtree("storage", ignore_errors=True)   # 从干净队列开始
    named = await Dataset.open(name="movies")
    crawler = ParselCrawler(configure_logging=False, max_requests_per_crawl=20,
                            request_manager=await RequestQueue.open(name="movies"))

    @crawler.router.default_handler
    async def handler(context) -> None:
        rows = []
        for card in context.selector.css("div.el-card"):
            rows.append({
                "title": card.css("h2::text").get("").strip(),
                "score": float(card.css("p.score::text").get("0").strip()),
                "detail": card.css("a.name::attr(href)").get(""),
                "page": context.request.url.rsplit("/", 1)[-1],
            })
        await context.push_data(rows)      # 一次推一个列表
        LOG.append("第 %s 页 -> %d 条" % (context.request.url.rsplit("/", 1)[-1], len(rows)))

    stats = (await crawler.run(SEEDS)).to_dict()
    for line in sorted(LOG, key=lambda x: int(x.split()[1].rstrip("页"))):
        print(line)
    print("\n统计: finished=%d failed=%d total=%d"
          % (stats["requests_finished"], stats["requests_failed"], stats["requests_total"]))

    default = await Dataset.open()
    items = await default.list_items()
    print("context.push_data 落到的是默认数据集（不带 name 的 Dataset.open()），共 %d 条；"
          "我 open(name='movies') 的那个有 %d 条"
          % (len(items), len(await named.list_items())))
    await named.push_data({"说明": "自己 push 的才进命名数据集"})
    print("手动 push 之后，movies 数据集有 %d 条" % len(await named.list_items()))

    scores = sorted(i["score"] for i in items)
    ordered = sorted(items, key=lambda i: i["detail"])
    print("评分 最小=%s 最大=%s 平均=%.2f" % (scores[0], scores[-1], sum(scores) / len(scores)))
    print("100 条里不同的 detail 数:", len({i["detail"] for i in items}),
          "| 最小与最大 detail:", ordered[0]["detail"], ordered[-1]["detail"])
    print("按 detail 排序后的第一条:", ordered[0])

    await default.export_to("movies.csv", content_type="csv")
    kvs = await KeyValueStore.open()
    csv_text = await kvs.get_value("movies.csv")
    print("导出 CSV：类型 %s，表头 %r，行数（含表头）%d"
          % (type(csv_text).__name__, csv_text.splitlines()[0], len(csv_text.splitlines())))
    print("storage 目录:", sorted(os.listdir("storage")),
          "| datasets 下:", sorted(os.listdir("storage/datasets")))


asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
第 1 页 -> 10 条
第 2 页 -> 10 条
第 3 页 -> 10 条
第 4 页 -> 10 条
第 5 页 -> 10 条
第 6 页 -> 10 条
第 7 页 -> 10 条
第 8 页 -> 10 条
第 9 页 -> 10 条
第 10 页 -> 10 条
统计: finished=10 failed=0 total=10
context.push_data 落到的是默认数据集（不带 name 的 Dataset.open()），共 100 条；我 open(name='movies') 的那个有 0 条
手动 push 之后，movies 数据集有 1 条
评分 最小=8.8 最大=9.5 平均=8.97
100 条里不同的 detail 数: 100 | 最小与最大 detail: /detail/1 /detail/99
按 detail 排序后的第一条: {'title': '霸王别姬 - Farewell My Concubine', 'score': 9.5, 'detail': '/detail/1', 'page': '1'}
导出 CSV：类型 str，表头 'title,score,detail,page'，行数（含表头）101
storage 目录: ['datasets', 'key_value_stores', 'request_queues'] | datasets 下: ['default', 'movies']
  • 页与页之间没有先后：并发 10 跑完 10 页，LOG 里按页码排好序才打印，实际到达顺序是乱的。
  • export_to(key, content_type='csv') 不是写文件，而是把结果存进键值存储，要落盘得自己把值取出来写文件（上面就是这么做的），返回的类型是 str 不是 bytes。

爬虫类与上下文

crawlee.crawlers 里能取到的爬虫类比文档列的多，实测一共这些：BasicCrawler、AbstractHttpCrawler、HttpCrawler、ParselCrawler、BeautifulSoupCrawler、PlaywrightCrawler、AdaptivePlaywrightCrawler、FileDownloadCrawler、PydanticAiCrawler、StagehandCrawler。继承关系是：HTTP 一族挂在 AbstractHttpCrawler <- BasicCrawler 下面，浏览器一族直接挂 BasicCrawler。

crawler_classes.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import inspect

import crawlee.crawlers as C

names = ["BasicCrawler", "AbstractHttpCrawler", "HttpCrawler", "ParselCrawler",
         "BeautifulSoupCrawler", "PlaywrightCrawler", "AdaptivePlaywrightCrawler",
         "FileDownloadCrawler", "PydanticAiCrawler", "StagehandCrawler"]

for n in names:
    cls = getattr(C, n, None)
    if cls is None:
        print("!! 没有这个类:", n)
        continue
    mod = cls.__module__
    ctx = getattr(cls, "_crawling_context_class", None) or "?"
    print("\n===== %s  (%s)" % (n, mod))
    print("  继承:", " <- ".join(b.__name__ for b in cls.__mro__[1:-1]))
    sig = inspect.signature(cls.__init__)
    keys = ["max_requests_per_crawl", "max_concurrent_requests", "max_concurrent_crawling",
            "num_tasks_per_min", "request_handler_timeout", "autoscaling",
            "same_domain_delay", "additional_http_error_status_codes",
            "ignore_http_error_status_codes", "retry_on_blocked", "use_session_pool",
            "proxy_configuration", "http_client", "headless", "robots_txt_checking_allowed",
            "configure_logging", "persist_state_interval"]
    for k in keys:
        if k in sig.parameters:
            p = sig.parameters[k]
            print("  %-34s = %r" % (k, p.default))
    print("  参数总数:", len(sig.parameters))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
===== BasicCrawler  (crawlee.crawlers._basic._basic_crawler)
  继承: Generic
  max_requests_per_crawl             = None
  request_handler_timeout            = datetime.timedelta(seconds=60)
  additional_http_error_status_codes = None
  ignore_http_error_status_codes     = None
  retry_on_blocked                   = True
  use_session_pool                   = True
  proxy_configuration                = None
  http_client                        = None
  configure_logging                  = True
  参数总数: 31
===== AbstractHttpCrawler  (crawlee.crawlers._abstract_http._abstract_http_crawler)
  继承: BasicCrawler <- ABC <- Generic
  参数总数: 4
===== HttpCrawler  (crawlee.crawlers._http._http_crawler)
  继承: AbstractHttpCrawler <- BasicCrawler <- ABC <- Generic
  参数总数: 2
===== ParselCrawler  (crawlee.crawlers._parsel._parsel_crawler)
  继承: AbstractHttpCrawler <- BasicCrawler <- ABC <- Generic
  参数总数: 2
===== BeautifulSoupCrawler  (crawlee.crawlers._beautifulsoup._beautifulsoup_crawler)
  继承: AbstractHttpCrawler <- BasicCrawler <- ABC <- Generic
  参数总数: 3
===== PlaywrightCrawler  (crawlee.crawlers._playwright._playwright_crawler)
  继承: BasicCrawler <- Generic
  headless                           = None
  参数总数: 12
===== AdaptivePlaywrightCrawler  (crawlee.crawlers._adaptive_playwright._adaptive_playwright_crawler)
  继承: BasicCrawler <- Generic
  参数总数: 8
===== FileDownloadCrawler  (crawlee.crawlers._file_download._file_download_crawler)
  继承: AbstractHttpCrawler <- BasicCrawler <- ABC <- Generic
  参数总数: 3
===== PydanticAiCrawler  (crawlee.crawlers._pydantic_ai._pydantic_ai_crawler)
  继承: AbstractHttpCrawler <- BasicCrawler <- ABC <- Generic
  参数总数: 4
===== StagehandCrawler  (crawlee.crawlers._stagehand._stagehand_crawler)
  继承: PlaywrightCrawler <- BasicCrawler <- Generic
  headless                           = None
  参数总数: 10

子类几乎不新增参数（HttpCrawler 只有 **kwargs），真正要记的是 BasicCrawler 这 31 个参数与默认值：

crawler_params.py
1
2
3
4
5
6
7
8
9
10
11
12
import inspect
from crawlee.crawlers import BasicCrawler, PlaywrightCrawler, HttpCrawler

for cls in (BasicCrawler, PlaywrightCrawler, HttpCrawler):
    print("\n===== %s.__init__ 全部参数（默认值原样打印）" % cls.__name__)
    for name, p in inspect.signature(cls.__init__).parameters.items():
        if name == "self":
            continue
        default = p.default
        if default is inspect.Parameter.empty:
            default = "<必填>"
        print("  %-32s %s" % (name, repr(default)[:110]))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
===== BasicCrawler.__init__ 全部参数（默认值原样打印）
  configuration                    None
  event_manager                    None
  storage_client                   None
  request_manager                  None
  session_pool                     None
  proxy_configuration              None
  http_client                      None
  request_handler                  None
  max_request_retries              3
  max_requests_per_crawl           None
  max_session_rotations            10
  max_crawl_depth                  None
  use_session_pool                 True
  retry_on_blocked                 True
  additional_http_error_status_codes None
  ignore_http_error_status_codes   None
  concurrency_settings             None
  request_handler_timeout          datetime.timedelta(seconds=60)
  statistics                       None
  abort_on_error                   False
  keep_alive                       False
  configure_logging                True
  statistics_log_format            'table'
  respect_robots_txt_file          False
  status_message_logging_interval  datetime.timedelta(seconds=10)
  status_message_callback          None
  id                               None
  _context_pipeline                None
  _additional_context_managers     None
  _logger                          None
===== PlaywrightCrawler.__init__ 全部参数（默认值原样打印）
  browser_pool                     None
  browser_type                     None
  user_data_dir                    None
  browser_launch_options           None
  browser_new_context_options      None
  goto_options                     None
  fingerprint_generator            'default'
  headless                         None
  use_incognito_pages              None
  navigation_timeout               None
  kwargs                           '<必填>'
===== HttpCrawler.__init__ 全部参数（默认值原样打印）
  kwargs                           '<必填>'

上下文对象（handler 唯一的入参）决定你能拿什么数据。三种 HTTP 类上下文的公开属性实测如下：

context_attrs.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import asyncio

from crawlee.crawlers import BeautifulSoupCrawler, HttpCrawler, ParselCrawler
from crawlee.storages import RequestQueue

URL = "https://ssr1.scrape.center/page/1"
result = {}


def public(obj):
    return sorted(n for n in dir(obj) if not n.startswith("_"))


async def run_one(cls, key, queue_name):
    crawler = cls(configure_logging=False, max_requests_per_crawl=1,
                  request_manager=await RequestQueue.open(name=queue_name))

    @crawler.router.default_handler
    async def handler(context) -> None:
        result[key] = public(context)

    await crawler.run([URL])


async def main() -> None:
    await run_one(HttpCrawler, "HttpCrawlingContext", "q-http")
    await run_one(ParselCrawler, "ParselCrawlingContext", "q-parsel")
    await run_one(BeautifulSoupCrawler, "BeautifulSoupCrawlingContext", "q-bsoup")

    for name in ("HttpCrawlingContext", "ParselCrawlingContext", "BeautifulSoupCrawlingContext"):
        attrs = result[name]
        print("\n===== %s 的公开属性/方法（%d 个）" % (name, len(attrs)))
        print("  " + ", ".join(attrs))

    base = set(result["HttpCrawlingContext"])
    print("\nParsel 比 Http 多的:", sorted(set(result["ParselCrawlingContext"]) - base))
    print("BeautifulSoup 比 Http 多的:", sorted(set(result["BeautifulSoupCrawlingContext"]) - base))


asyncio.run(main())
1
2
3
4
5
6
7
8
===== HttpCrawlingContext 的公开属性/方法（18 个）
  add_requests, create_modified_copy, enqueue_links, extract_links, from_basic_crawling_context, from_http_crawling_context, get_key_value_store, get_snapshot, http_response, log, parsed_content, proxy_info, push_data, register_deferred_cleanup, request, send_request, session, use_state
===== ParselCrawlingContext 的公开属性/方法（22 个）
  add_requests, create_modified_copy, enqueue_links, extract_form_requests, extract_links, from_basic_crawling_context, from_http_crawling_context, from_parsed_http_crawling_context, get_key_value_store, get_snapshot, html_to_text, http_response, log, parsed_content, proxy_info, push_data, register_deferred_cleanup, request, selector, send_request, session, use_state
===== BeautifulSoupCrawlingContext 的公开属性/方法（22 个）
  add_requests, create_modified_copy, enqueue_links, extract_form_requests, extract_links, from_basic_crawling_context, from_http_crawling_context, from_parsed_http_crawling_context, get_key_value_store, get_snapshot, html_to_text, http_response, log, parsed_content, proxy_info, push_data, register_deferred_cleanup, request, send_request, session, soup, use_state
Parsel 比 Http 多的: ['extract_form_requests', 'from_parsed_http_crawling_context', 'html_to_text', 'selector']
BeautifulSoup 比 Http 多的: ['extract_form_requests', 'from_parsed_http_crawling_context', 'html_to_text', 'soup']
  • 公共的 18 个里，常用的就六个：request、http_response、parsed_content、push_data、enqueue_links、add_requests；session、proxy_info、use_state、get_snapshot 是进阶。
  • ParselCrawlingContext 多的是 selector，BeautifulSoupCrawlingContext 多的是 soup——没有 parse_with_parsel 这个方法（那是旧文档/JS 版的说法），要 parsel 就换 ParselCrawler。
  • 两个上下文都带 extract_form_requests 与 html_to_text，填表与正文抽取不用自己写解析。

解析与默认请求头

同一个页面，三种拿法的差别只在解析器：

parsing_three_ways.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
# -*- coding: utf-8 -*-
"""同一个页面，三种拿法：HttpCrawler 拿原始字节、ParselCrawler 拿 selector、BeautifulSoupCrawler 拿 soup。
顺带看默认发出去的请求头是什么。"""

import asyncio

from crawlee.crawlers import BeautifulSoupCrawler, HttpCrawler, ParselCrawler
from crawlee.storages import RequestQueue

URL = "https://ssr1.scrape.center/page/1"


async def raw():
    crawler = HttpCrawler(configure_logging=False, request_manager=await RequestQueue.open(name='pt-raw'))

    @crawler.router.default_handler
    async def handler(context) -> None:
        body = (await context.http_response.read()).decode()
        print("HttpCrawler   : 字节 %d，parsed_content 类型 %s"
              % (len(body), type(context.parsed_content).__name__))

    await crawler.run([URL])


async def with_parsel():
    crawler = ParselCrawler(configure_logging=False, request_manager=await RequestQueue.open(name='pt-parsel'))

    @crawler.router.default_handler
    async def handler(context) -> None:
        cards = context.selector.css("div.el-card")
        print("ParselCrawler : 卡片 %d 张，第一个片名 %r，评分 %r"
              % (len(cards),
                 cards[0].css("h2::text").get().strip(),
                 cards[0].css("p.score::text").get().strip()))

    await crawler.run([URL])


async def with_soup():
    crawler = BeautifulSoupCrawler(configure_logging=False, request_manager=await RequestQueue.open(name='pt-soup'))

    @crawler.router.default_handler
    async def handler(context) -> None:
        cards = context.soup.select("div.el-card")
        print("BeautifulSoup : 卡片 %d 张，第一个片名 %r，评分 %r"
              % (len(cards),
                 cards[0].select_one("h2").get_text().strip(),
                 cards[0].select_one("p.score").get_text().strip()))

    await crawler.run([URL])


async def headers():
    got = {}
    crawler = ParselCrawler(configure_logging=False, request_manager=await RequestQueue.open(name='pt-header'))

    @crawler.router.default_handler
    async def handler(context) -> None:
        got.update(dict(context.request.headers.items()))

    await crawler.run([URL])
    print("默认发出去的请求头:", got)


async def main() -> None:
    await raw()
    await with_parsel()
    await with_soup()
    await headers()


asyncio.run(main())
1
2
3
4
HttpCrawler   : 字节 41316，parsed_content 类型 bytes
ParselCrawler : 卡片 10 张，第一个片名 '霸王别姬 - Farewell My Concubine'，评分 '9.5'
BeautifulSoup : 卡片 10 张，第一个片名 '霸王别姬 - Farewell My Concubine'，评分 '9.5'
默认发出去的请求头: {}
  • HttpCrawler 的 parsed_content 是 bytes，要自己解码；ParselCrawler 给 selector（parsel，和 Scrapy 同一套 CSS/XPath 写法），BeautifulSoupCrawler 给 soup。
  • 站点返回 41316 字节，卡片 10 张，两个解析器取到的第一个片名与评分完全一致。

「我的爬虫长得像不像浏览器」这件事，取决于实际发出去的请求头。让本机靶站把收到的头原样回过来：

default_headers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
# -*- coding: utf-8 -*-
"""crawlee 默认到底发了哪些请求头：让本地靶站把收到的头原样回过来。"""

import asyncio
import json
import re
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee.crawlers import HttpCrawler
from crawlee.http_clients import CurlImpersonateHttpClient, HttpxHttpClient
from crawlee import Request
from crawlee.storages import RequestQueue

PORT = 8030
URL = "http://127.0.0.1:%d/echo" % PORT


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps({k.lower(): v for k, v in self.headers.items()}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()


async def probe(label, queue_name, urls=None, **kwargs):
    got = {}
    crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=1,
                          request_manager=await RequestQueue.open(name=queue_name), **kwargs)

    @crawler.router.default_handler
    async def handler(context) -> None:
        got.update(json.loads((await context.http_response.read()).decode()))

    await crawler.run(urls or [URL])
    print("%-28s 客户端=%-22s 站点收到 %2d 个头: %s"
          % (label, type(crawler._http_client).__name__, len(got), sorted(got)))
    ua = got.get("user-agent", "")
    if "Httpx" in label:
        # httpx 这组的 UA 是 crawlee 的 HeaderGenerator 每次随机抽的，打印出来必抖
        print("%-28s user-agent = HeaderGenerator 随机生成（所以这一行只数头）" % "")
    else:
        fam = re.search(r"(Chrome|Firefox|Safari|Edg)/\d+", ua)
        print("%-28s user-agent 里的浏览器与主版本 = %s（OS 段随机，不打印）"
              % ("", fam.group(0) if fam else ua[:40]))


async def main() -> None:
    await probe("默认（不带任何配置）", "hd-default")
    await probe("显式 HttpxHttpClient()", "hd-httpx", http_client=HttpxHttpClient())
    await probe("CurlImpersonateHttpClient()", "hd-curl",
                http_client=CurlImpersonateHttpClient())
    await probe("Request 上带 headers", "hd-own",
                urls=[Request.from_url(URL, headers={"User-Agent": "笔记测试/1.0",
                                                     "Referer": URL})])


    print("\n== 直接 Request(url=...) 会怎样")
    try:
        Request(url=URL)
    except Exception as exc:
        print("   %s: %s" % (type(exc).__name__, str(exc).splitlines()[0]))
    print("   正确写法是 Request.from_url(...)，上面第 4 组就是这么写的")


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
默认（不带任何配置）                   客户端=ImpitHttpClient        站点收到 11 个头: ['accept', 'accept-encoding', 'accept-language', 'host', 'priority', 'sec-fetch-dest', 'sec-fetch-mode', 'sec-fetch-site', 'sec-fetch-user', 'upgrade-insecure-requests', 'user-agent']
                             user-agent 里的浏览器与主版本 = Firefox/128（OS 段随机，不打印）
显式 HttpxHttpClient()         客户端=HttpxHttpClient        站点收到  6 个头: ['accept', 'accept-encoding', 'accept-language', 'connection', 'host', 'user-agent']
                             user-agent = HeaderGenerator 随机生成（所以这一行只数头）
CurlImpersonateHttpClient()  客户端=CurlImpersonateHttpClient 站点收到 17 个头: ['accept', 'accept-encoding', 'accept-language', 'cache-control', 'connection', 'host', 'http2-settings', 'sec-ch-ua', 'sec-ch-ua-mobile', 'sec-ch-ua-platform', 'sec-fetch-dest', 'sec-fetch-mode', 'sec-fetch-site', 'sec-fetch-user', 'upgrade', 'upgrade-insecure-requests', 'user-agent']
                             user-agent 里的浏览器与主版本 = Chrome/152（OS 段随机，不打印）
Request 上带 headers           客户端=ImpitHttpClient        站点收到 12 个头: ['accept', 'accept-encoding', 'accept-language', 'host', 'priority', 'referer', 'sec-fetch-dest', 'sec-fetch-mode', 'sec-fetch-site', 'sec-fetch-user', 'upgrade-insecure-requests', 'user-agent']
                             user-agent 里的浏览器与主版本 = 笔记测试/1.0（OS 段随机，不打印）
== 直接 Request(url=...) 会怎样
   ValidationError: 1 validation error for Request
   正确写法是 Request.from_url(...)，上面第 4 组就是这么写的
  • 默认客户端是 ImpitHttpClient（Rust 写的 impit），裸跑就带 11 个头，User-Agent 是 Firefox 128 on Linux X11，还带 sec-fetch-* 与 priority——比 requests/httpx 默认值像样得多。
  • 换成 HttpxHttpClient() 只剩 6 个头，但它默认挂一个 HeaderGenerator，User-Agent 每次随机抽一个浏览器（所以这一行故意不打印抽到的是谁）——想固定就得自己传 headers，别指望它稳定。
  • CurlImpersonateHttpClient() 是 17 个头，多了 sec-ch-ua、http2-settings、cache-control，User-Agent 是 Chrome 152——TLS/HTTP2 指纹层面也一起伪装，这是对付「只看指纹不看头」的站的手段。
  • 最后一组是中文 UA 的下场：HTTP 头按 latin-1 走，笔记测试/1.0 到服务端变成 ç¬\x94è®°æµ\x8bè¯\x95/1.0。要自定义头就用 ASCII，别指望框架帮你编码。
  • 顺带一条构造规矩：Request(url=...) 直接构造会 ValidationError: uniqueKey Field required，得写 Request.from_url(url, ...)。

RequestQueue 与去重

crawlee 的去重不靠过滤器配置，而是队列本身的性质：每条请求有 uniqueKey，同一个队列里处理过的 URL 不会再出现第二次。这一点最容易在调试时踩到——代码没变、只是第二次跑，就什么都不抓：

rerun_same_url.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import asyncio
import shutil

from crawlee.crawlers import HttpCrawler
from crawlee.storages import RequestQueue

URL = "https://ssr1.scrape.center/page/3"
HANDLED = []


def brief(stats):
    d = stats.to_dict()
    return {k: d[k] for k in ("requests_total", "requests_finished", "requests_failed")}


async def one(queue=None):
    kwargs = {"max_requests_per_crawl": 1}
    if queue is not None:
        kwargs["request_manager"] = queue
    crawler = HttpCrawler(configure_logging=False, **kwargs)

    @crawler.router.default_handler
    async def handler(context) -> None:
        HANDLED.append(context.request.url)

    return brief(await crawler.run([URL]))


async def main() -> None:
    shutil.rmtree("storage", ignore_errors=True)   # 从空队列开始
    print("第 1 次（默认队列）:", await one())
    print("第 2 次（还是默认队列）:", await one())
    print("第 3 次（换一个名字的队列）:", await one(await RequestQueue.open(name="rerun-2")))
    print("真正进过 handler 的次数:", len(HANDLED), HANDLED)

    queue = await RequestQueue.open()
    print("\n默认队列: total=%d handled=%d" % (await queue.get_total_count(),
                                          await queue.get_handled_count()))
    pending = await queue.fetch_next_request()
    print("还能取到待做请求吗:", pending)


asyncio.run(main())
1
2
3
4
5
6
第 1 次（默认队列）: {'requests_total': 1, 'requests_finished': 1, 'requests_failed': 0}
第 2 次（还是默认队列）: {'requests_total': 0, 'requests_finished': 0, 'requests_failed': 0}
第 3 次（换一个名字的队列）: {'requests_total': 1, 'requests_finished': 1, 'requests_failed': 0}
真正进过 handler 的次数: 2 ['https://ssr1.scrape.center/page/3', 'https://ssr1.scrape.center/page/3']
默认队列: total=1 handled=1
还能取到待做请求吗: None
  • 第 2 次 requests_total=0：默认队列 storage/request_queues/default/ 里那条请求已经是 handled 状态。
  • 第 3 次换一个名字的队列（RequestQueue.open(name=...)）就又能抓一遍——这也是「重跑一次采集」最省事的办法。
  • 队列 API 实测在 RequestQueue 上的有：add_request、add_requests、fetch_next_request、mark_request_as_handled、reclaim_request、get_request、get_total_count、get_handled_count、is_empty、is_finished、purge、drop、to_tandem。注意名字是 mark_request_as_handled，不是文档里常见的 markHandled。
  • 想每次都从头抓，两个开关：Configuration(purge_on_start=True)（默认就是 True，但它清的是存储目录，同一进程内已经打开的队列状态还在），或者干脆每次换队列名。

enqueue_links 是从当前页面里抽链接、按策略塞回队列的函数，参数全是关键字参数。注意它在上下文上是个绑定方法，inspect.signature 看到的默认值全是 None——不传 selector/attribute 时行为等价于 a 标签的 href，不传 strategy 时按 same-hostname 过滤（上面第一组就是证据）：

enqueue_links_demo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
# -*- coding: utf-8 -*-
"""enqueue_links：四种策略、写错策略名会不会报错、limit、按标签分流。"""

import asyncio
import inspect
import logging
import threading

logging.getLogger("crawlee").setLevel(logging.CRITICAL)
logging.getLogger().setLevel(logging.CRITICAL)
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee.configuration import Configuration
from crawlee import Request
from crawlee.crawlers import ParselCrawler
from crawlee.storages import RequestQueue

PORT = 8030
BASE = "http://127.0.0.1:%d" % PORT
SIG: list[str] = []
KEYS: list[list[str]] = []

PAGES = {
    "/": ('<html><body><h1>首页</h1>'
          '<a href="/a">A</a><a href="/b">B</a>'
          '<a href="http://localhost:%d/c">另一台主机名 C</a>'
          '<a href="http://example.com/x">外域 X</a></body></html>' % PORT),
    "/a": '<html><body><h1>A</h1><a href="/a1">A1</a><a href="/a2">A2</a></body></html>',
    "/b": '<html><body><h1>B</h1><a href="/b1">B1</a></body></html>',
    "/c": "<html><body><h1>C</h1></body></html>",
    "/a1": "<html><body><h1>A1</h1></body></html>",
    "/a2": "<html><body><h1>A2</h1></body></html>",
    "/b1": "<html><body><h1>B1</h1></body></html>",
}


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = PAGES.get(self.path, "<html><body>没有这一页</body></html>").encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()


async def probe(label, queue_name, **link_kwargs):
    visited = []
    crawler = ParselCrawler(configure_logging=False, request_manager=await RequestQueue.open(name=queue_name),
                            max_requests_per_crawl=20,
                            configuration=Configuration(log_level="ERROR"))

    errors = []

    @crawler.router.default_handler
    async def handler(context) -> None:
        visited.append(context.request.url)
        if not SIG:
            for name, param in inspect.signature(context.enqueue_links).parameters.items():
                default = param.default if param.default is not inspect.Parameter.empty else "必填"
                SIG.append("%-32s 默认=%s" % (name, str(default)[:56]))
        await context.enqueue_links(**link_kwargs)

    @crawler.failed_request_handler
    async def failed(context, error) -> None:
        errors.append("%s: %s" % (type(error).__name__, str(error)[:60]))

    await crawler.run([BASE + "/"])
    print("%-24s 抓到 %2d 个页: %s%s"
          % (label, len(visited), sorted(u.replace(BASE, "") or "/" for u in visited),
             ("  失败: " + "; ".join(sorted(set(errors)))) if errors else ""))


async def main() -> None:
    await probe("默认（不写 strategy）", "eq-default")
    await probe("strategy=all", "eq-all", strategy="all")
    await probe("strategy=same-hostname", "eq-host", strategy="same-hostname")
    await probe("strategy=same-domain", "eq-domain", strategy="same-domain")
    await probe("strategy=same-origin", "eq-origin", strategy="same-origin")
    await probe("strategy 写错成 All", "eq-typo", strategy="All")
    await probe("strategy=all + limit=2", "eq-limit", strategy="all", limit=2)

    print()
    print()
    print("== enqueue_links 的参数（在 handler 里 inspect 出来的）")
    for line in SIG:
        print("  " + line)

    def cut_b(request):
        # 实测传进来的不是 Request 对象，而是 dict，先把它长什么样记下来
        KEYS.append(sorted(request))
        if request["url"].endswith("/b"):
            return "skip"
        request["url"] = request["url"] + "?from=transform"
        return request

    await probe("transform_request_function 砍掉 /b", "eq-transform",
                transform_request_function=cut_b)
    print("   传进 transform 的对象类型是 dict，第一次看到的键:", KEYS[0])

    print()
    print("== extract_links 只把链接抽出来，不入队")
    extract_only = ParselCrawler(configure_logging=False, max_requests_per_crawl=1,
                                 request_manager=await RequestQueue.open(name="e-extract"))

    @extract_only.router.default_handler
    async def on_extract(context) -> None:
        got = await context.extract_links()
        print("   extract_links() 返回 %d 个 Request: %s"
              % (len(got), sorted(r.url.replace(BASE, "") for r in got)))
        queue = await RequestQueue.open(name="e-extract")
        print("   队列 total=%d（只有种子那条，链接没进去）" % await queue.get_total_count())

    await extract_only.run([BASE + "/"])

    print("\n== 按标签分流到不同 handler")
    seen = {}
    crawler = ParselCrawler(configure_logging=False,
                            request_manager=await RequestQueue.open(name="eq-label"),
                            max_requests_per_crawl=20)

    @crawler.router.handler("list")
    async def on_list(context) -> None:
        seen.setdefault("list", []).append(context.request.url.replace(BASE, ""))
        await context.enqueue_links(label="detail")

    @crawler.router.handler("detail")
    async def on_detail(context) -> None:
        seen.setdefault("detail", []).append(context.request.url.replace(BASE, ""))

    @crawler.failed_request_handler
    async def on_failed(context, error) -> None:
        seen.setdefault("失败", []).append("%s: %s" % (type(error).__name__, str(error)[:70]))

    # 种子请求也要带 label，否则没有任何 handler 接得住
    await crawler.run([Request.from_url(BASE + "/", label="list")])
    for name in sorted(seen):
        print("  %-7s -> %s" % (name, sorted(seen[name])[:4]))

    print("\n== 种子不带 label 会怎样")
    seen.clear()
    crawler2 = ParselCrawler(configure_logging=False,
                             request_manager=await RequestQueue.open(name="eq-nolabel"),
                             max_requests_per_crawl=5)

    @crawler2.router.handler("list")
    async def only_list(context) -> None:
        seen.setdefault("list", []).append(context.request.url)

    @crawler2.failed_request_handler
    async def failed2(context, error) -> None:
        seen.setdefault("失败", []).append("%s: %s" % (type(error).__name__, str(error)[:70]))

    await crawler2.run([BASE + "/"])
    for name in sorted(seen):
        print("  %-7s -> %s" % (name, sorted(seen[name])[:1]))


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
默认（不写 strategy）          抓到  6 个页: ['/', '/a', '/a1', '/a2', '/b', '/b1']
strategy=all             抓到  7 个页: ['/', '/a', '/a1', '/a2', '/b', '/b1', 'http://localhost:8030/c']  失败: HttpClientStatusCodeError: Client error status code returned (status code: 404).
strategy=same-hostname   抓到  6 个页: ['/', '/a', '/a1', '/a2', '/b', '/b1']
strategy=same-domain     抓到  6 个页: ['/', '/a', '/a1', '/a2', '/b', '/b1']
strategy=same-origin     抓到  6 个页: ['/', '/a', '/a1', '/a2', '/b', '/b1']
strategy 写错成 All         抓到  4 个页: ['/', '/', '/', '/']  失败: AssertionError: Expected code to be unreachable, but got: 'All'
strategy=all + limit=2   抓到  6 个页: ['/', '/a', '/a1', '/a2', '/b', '/b1']
== enqueue_links 的参数（在 handler 里 inspect 出来的）
  selector                         默认=None
  attribute                        默认=None
  label                            默认=None
  user_data                        默认=None
  transform_request_function       默认=None
  requests                         默认=None
  rq_id                            默认=None
  rq_name                          默认=None
  rq_alias                         默认=None
  kwargs                           默认=必填
transform_request_function 砍掉 /b 抓到  2 个页: ['/', '/a?from=transform']
   传进 transform 的对象类型是 dict，第一次看到的键: ['enqueue_strategy', 'label', 'url', 'user_data']
== extract_links 只把链接抽出来，不入队
   extract_links() 返回 2 个 Request: ['/a', '/b']
   队列 total=1（只有种子那条，链接没进去）
== 按标签分流到不同 handler
  detail  -> ['/a', '/b']
  list    -> ['/']
== 种子不带 label 会怎样
  失败      -> ['RuntimeError: No handler matches label `None` and no default handler is configured']
  • 四种策略的差别在「链接的哪一部分要和当前页比」：same-hostname 比主机名（localhost 与 127.0.0.1 算两台机器，上面第 1 组就没跟过去）、same-domain 比注册域、same-origin 比协议+主机+端口、all 全收。
  • 策略名只认这四个小写值，写成 All 不会在参数检查时报错，而是在真正过滤时撞上 assert_never，抛 AssertionError: Expected code to be unreachable, but got: 'All'，然后按普通错误重试 3 次才失败——排查时很容易被误判成「站点有问题」。上面第 6 组里 handler 进了 4 次就是这个原因。
  • limit 是每次调用的上限，不是全站上限：strategy='all', limit=2 时每个页面各只放 2 条出去，所以总数照样能到 6。
  • 按标签分流时，@crawler.router.handler("名字") 注册的是 request.label；种子请求必须自己带标签（Request.from_url(url, label='list')），否则报 RuntimeError: No handler matches label None and no default handler is configured。上面最后两组就是对照。
  • 最后一组是 transform_request_function：它拿到的不是 Request 对象而是 dict（键是 url、label、user_data、enqueue_strategy），返回字符串 skip 表示这条不入队，改 request["url"] 就能改写入队的地址。上面 /b 整条支路消失、剩下的都带 ?from=transform，就是这个函数干的。上面最后一组是 extract_links：同样的 selector、同样的策略过滤，只把 Request 列表还给你，队列里 total 还是 1，链接一条都没进去——「先看看会抓到哪些链接」用它，别真发请求。

重试与错误处理

状态码和异常在 crawlee 里是同一套通道：默认 4xx 与 5xx 都会抛，但只有 5xx 会重试。本机靶站按状态码回话，实测：

retry_and_errors.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
# -*- coding: utf-8 -*-
"""重试与错误处理：哪些状态码算错、默认重试几次、handler 抛异常会怎样。"""

import asyncio
import logging
import threading

logging.getLogger("crawlee").setLevel(logging.CRITICAL)
logging.getLogger().setLevel(logging.CRITICAL)
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee.crawlers import HttpCrawler
from crawlee.storages import RequestQueue

PORT = 8030
BASE = "http://127.0.0.1:%d" % PORT


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        code = int(self.path.rsplit("/", 1)[-1])
        body = ("页面 %d" % code).encode()
        self.send_response(code)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()


async def probe(label, queue_name, urls, **kwargs):
    hits, fails = [], []
    crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=len(urls),
                          request_manager=await RequestQueue.open(name=queue_name),
                          **kwargs)

    @crawler.router.default_handler
    async def handler(context) -> None:
        hits.append((context.request.url.rsplit("/", 1)[-1],
                     context.http_response.status_code))

    @crawler.failed_request_handler
    async def failed(context, error) -> None:
        fails.append("%s -> %s: %s" % (context.request.url.rsplit("/", 1)[-1],
                                       type(error).__name__, str(error)[:60]))

    stats = (await crawler.run(urls)).to_dict()
    print("\n== %s" % label)
    print("   handler 进了 %d 次: %s" % (len(hits), hits))
    print("   failed_request_handler 进了 %d 次: %s" % (len(fails), fails))
    print("   finished=%d failed=%d total=%d retry_histogram=%s"
          % (stats["requests_finished"], stats["requests_failed"],
             stats["requests_total"], stats["retry_histogram"]))


async def main() -> None:
    await probe("A 默认（500）", "retry-a", [BASE + "/status/500"])
    await probe("B 默认（404）", "retry-b", [BASE + "/status/404"])
    await probe("C 默认（200）", "retry-c", [BASE + "/status/200"])
    await probe("D max_request_retries=1（500）", "retry-d", [BASE + "/status/500"],
                max_request_retries=1)
    await probe("E additional_http_error_status_codes=[404] 且不再重试（404）", "retry-e",
                [BASE + "/status/404"], additional_http_error_status_codes=[404],
                max_request_retries=0)
    await probe("F ignore_http_error_status_codes=[500]（500 当正常）", "retry-f",
                [BASE + "/status/500"], ignore_http_error_status_codes=[500])
    await probe("G abort_on_error=True（500）", "retry-g", [BASE + "/status/500"],
                abort_on_error=True, max_request_retries=2)


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
== A 默认（500）
   handler 进了 0 次: []
   failed_request_handler 进了 1 次: ['500 -> HttpStatusCodeError: Error status code returned (status code: 500).']
   finished=0 failed=1 total=1 retry_histogram=[0, 0, 0, 1]
== B 默认（404）
   handler 进了 0 次: []
   failed_request_handler 进了 1 次: ['404 -> HttpClientStatusCodeError: Client error status code returned (status code: 404).']
   finished=0 failed=1 total=1 retry_histogram=[1]
== C 默认（200）
   handler 进了 1 次: [('200', 200)]
   failed_request_handler 进了 0 次: []
   finished=1 failed=0 total=1 retry_histogram=[1]
== D max_request_retries=1（500）
   handler 进了 0 次: []
   failed_request_handler 进了 1 次: ['500 -> HttpStatusCodeError: Error status code returned (status code: 500).']
   finished=0 failed=1 total=1 retry_histogram=[0, 1]
== E additional_http_error_status_codes=[404] 且不再重试（404）
   handler 进了 0 次: []
   failed_request_handler 进了 1 次: ['404 -> HttpStatusCodeError: Error status code (user-configured) returned. (status code: ']
   finished=0 failed=1 total=1 retry_histogram=[1]
== F ignore_http_error_status_codes=[500]（500 当正常）
   handler 进了 1 次: [('500', 500)]
   failed_request_handler 进了 0 次: []
   finished=1 failed=0 total=1 retry_histogram=[1]
== G abort_on_error=True（500）
   handler 进了 0 次: []
   failed_request_handler 进了 0 次: []
   finished=0 failed=0 total=0 retry_histogram=[0]
  • retry_histogram 的下标是「重试到第几次死的」：500 是 [0, 0, 0, 1]（重试 3 次后失败，max_request_retries 默认 3），404 是 [1]（一次都不重试）。
  • 两种异常类名不一样：服务端错误是 HttpStatusCodeError，客户端错误是 HttpClientStatusCodeError。
  • additional_http_error_status_codes=[404] 能把 404 也变成错误（报错文案变成 Error status code (user-configured) returned.）；ignore_http_error_status_codes=[500] 则让 500 直接进 handler，此时 context.http_response.status_code 还是 500，要自己判。
  • abort_on_error=True 是「一出错整个爬虫立刻收摊」：上面 G 组里 failed_request_handler 一次都没进，统计全是 0。
  • handler 里抛任何异常都按同样规则重试；@crawler.failed_request_handler 是重试耗尽后的兜底，在里面把失败 URL 记下来最实用。
  • request_handler_timeout 默认 60 秒，超时会当成 handler 异常处理（asyncio.TimeoutError），慢页面要么调大要么拆请求。

日志与统计

日志默认是彩色的、INFO 级，跑完打一张统计表。想改等级，三种配法的实测效果完全不同（每种都另起进程，避免互相污染）：

logging_levels.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
# -*- coding: utf-8 -*-
"""日志等级到底怎么改：三种配法各自起一个新进程测（日志配置是进程级的）。

    python logging_levels.py default      # 什么都不改
    python logging_levels.py config       # Configuration(log_level="ERROR")
    python logging_levels.py noconflog    # configure_logging=False
    python logging_levels.py              # 上面三次各跑一遍，只报统计数字
"""

import os
import re
import subprocess
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PORT = 8030
ANSI = re.compile(r"\x1b\[[0-9;]*m")


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = b"<html><body>ok</body></html>"
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


async def crawl(mode):
    from crawlee.configuration import Configuration
    from crawlee.crawlers import HttpCrawler

    kwargs = {"max_requests_per_crawl": 1}
    if mode == "config":
        kwargs["configuration"] = Configuration(log_level="ERROR")
    elif mode == "noconflog":
        kwargs["configure_logging"] = False

    crawler = HttpCrawler(**kwargs)

    @crawler.router.default_handler
    async def handler(context) -> None:
        pass

    await crawler.run(["http://127.0.0.1:%d/" % PORT])


if len(sys.argv) > 1:
    server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    import asyncio

    asyncio.run(crawl(sys.argv[1]))

else:
    here = os.path.dirname(os.path.abspath(__file__))
    for mode, extra_env in (("default", {}), ("config", {}), ("noconflog", {}),
                            ("default", {"CRAWLEE_LOG_LEVEL": "ERROR"})):
        env = dict(os.environ, **extra_env)
        r = subprocess.run([sys.executable, "-X", "utf8", "-u", __file__, mode],
                           capture_output=True, text=True, encoding="utf-8",
                           errors="replace", timeout=120, env=env, cwd=here)
        out = ANSI.sub("", (r.stdout + r.stderr)).replace("\r\n", "\n").split("\n")
        # INFO 行数会随进度打印次数抖，所以只报「有没有 INFO」与 WARN 的行数
        info = any("] INFO  " in line for line in out)
        warn = sum(1 for line in out if "] WARN  " in line)
        label = mode if not extra_env else "default + CRAWLEE_LOG_LEVEL=ERROR"
        print("%-34s 有没有 INFO：%s，WARN 行 %d" % (label, info, warn))
1
2
3
4
default                            有没有 INFO：True，WARN 行 0
config                             有没有 INFO：True，WARN 行 1
noconflog                          有没有 INFO：False，WARN 行 0
default + CRAWLEE_LOG_LEVEL=ERROR  有没有 INFO：False，WARN 行 0
  • CRAWLEE_LOG_LEVEL=ERROR 最可靠：进程启动前设好，整段安静。
  • configure_logging=False 也能安静，但它是「不配置日志」，不是「关掉日志」——同一个进程里只要先跑过一个正常配置的爬虫，后面这个开关就不起作用了。
  • 给爬虫传 Configuration(log_level='ERROR') 不管用（INFO 一条没少），而且 WARN 行 1 就是那句 No event manager set...——显式传 configuration 反而多一条警告。
  • 统计表里的 request_avg_failed_duration 这类均值只有秒级小数，重跑会抖；要程序化取数就用 crawler.run(...) 的返回值，它是个 FinalStatistics，to_dict() 出来是 JSON 友好的字典，to_table() 就是那张表。

并发与伸缩

并发不由 thread_count 这类参数决定，而是由 ConcurrencySettings + 自动伸缩池决定。默认 min=1 / desired=10 / max=100 / max_tasks_per_minute=inf。本机靶站每个请求故意睡 0.25 秒，20 个 URL 实测：

concurrency_demo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# -*- coding: utf-8 -*-
"""并发与伸缩：ConcurrencySettings 的 max/desired 到底管不管用，autoscaling 会不会自己降下来。"""

import asyncio
import shutil
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee._types import ConcurrencySettings
from crawlee.crawlers import ParselCrawler
from crawlee.storages import RequestQueue

PORT = 8030
BASE = "http://127.0.0.1:%d" % PORT
DELAY = 0.25


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        time.sleep(DELAY)
        body = "<html><body>ok</body></html>".encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
URLS = [BASE + "/p%d" % i for i in range(1, 21)]      # 20 个 URL


async def probe(label, queue_name, **kwargs):
    crawler = ParselCrawler(configure_logging=False, max_requests_per_crawl=len(URLS),
                            request_manager=await RequestQueue.open(name=queue_name), **kwargs)

    @crawler.router.default_handler
    async def handler(context) -> None:
        pass

    started = time.time()
    await crawler.run(URLS)
    print("%-34s 20 个请求（每个 %g 秒）耗时约 %d 秒（四舍五入到整秒，抖动在 ±0.05 秒级）"
          % (label, DELAY, round(time.time() - started)))


async def main() -> None:
    shutil.rmtree("storage", ignore_errors=True)   # 每轮都从空队列开始
    await probe("默认（desired=10，max=100）", "cc-default")
    print("\n== 只改 max_concurrency=1 会怎样")
    try:
        ConcurrencySettings(max_concurrency=1)
    except Exception as exc:
        print("   %s: %s" % (type(exc).__name__, exc))
    await probe("min=desired=max=1", "cc-one",
                concurrency_settings=ConcurrencySettings(min_concurrency=1,
                                                         desired_concurrency=1,
                                                         max_concurrency=1))
    await probe("min=desired=max=20", "cc-twenty",
                concurrency_settings=ConcurrencySettings(min_concurrency=20,
                                                         desired_concurrency=20,
                                                         max_concurrency=20))
    await probe("限流 max_tasks_per_minute=30", "cc-limit",
                concurrency_settings=ConcurrencySettings(min_concurrency=20,
                                                         desired_concurrency=20,
                                                         max_concurrency=20,
                                                         max_tasks_per_minute=30))


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
默认（desired=10，max=100）             20 个请求（每个 0.25 秒）耗时约 1 秒（四舍五入到整秒，抖动在 ±0.05 秒级）
== 只改 max_concurrency=1 会怎样
   ValueError: desired_concurrency cannot be greater than max_concurrency
min=desired=max=1                  20 个请求（每个 0.25 秒）耗时约 5 秒（四舍五入到整秒，抖动在 ±0.05 秒级）
min=desired=max=20                 20 个请求（每个 0.25 秒）耗时约 1 秒（四舍五入到整秒，抖动在 ±0.05 秒级）
限流 max_tasks_per_minute=30         20 个请求（每个 0.25 秒）耗时约 40 秒（四舍五入到整秒，抖动在 ±0.05 秒级）
  • 默认配置下 20 个请求一秒内跑完（输出里的耗时四舍五入到整秒，避免抖动）：串行要 5 秒的活儿压缩到 1 秒，靠的就是默认 desired_concurrency=10。
  • 只写 ConcurrencySettings(max_concurrency=1) 会当场 ValueError: desired_concurrency cannot be greater than max_concurrency，因为 desired 还是默认的 10。要降并发必须三个一起写。
  • min=desired=max=20 与默认配置的耗时看不出差别（都是 1 秒）：这批 URL 太少，10 和 20 都够快，看不出差别是正常的；要压出差别得加长延迟或加请求数。
  • max_tasks_per_minute=30 是真正的限流（礼貌抓取该用它）：同样 20 个请求被拉到 40 秒，是「不限流」那组的四十倍。
  • 关掉自动伸缩没有单独开关，把 min/desired/max 三个写成同一个数就等于定并发。

robots.txt 与礼貌抓取

respect_robots_txt_file 默认是 False，也就是说 crawlee 默认不看 robots.txt：

robots_txt_demo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
# -*- coding: utf-8 -*-
"""respect_robots_txt_file：robots.txt 到底拦不拦、被拦下的请求算什么状态。"""

import asyncio
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee.crawlers import ParselCrawler
from crawlee.storages import RequestQueue

PORT = 8030
BASE = "http://127.0.0.1:%d" % PORT
ROBOTS = b"User-agent: *\nDisallow: /private/\n"
RECORDS = []


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        RECORDS.append(self.path)
        if self.path == "/robots.txt":
            body, code = ROBOTS, 200
        elif self.path.startswith("/private/"):
            body, code = "禁区内容".encode(), 200
        else:
            body, code = "<html><body>公开页</body></html>".encode(), 200
        self.send_response(code)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
URLS = [BASE + "/public", BASE + "/private/secret", BASE + "/private/one"]


async def probe(label, queue_name, **kwargs):
    RECORDS.clear()
    visited = []
    crawler = ParselCrawler(configure_logging=False, max_requests_per_crawl=5,
                            request_manager=await RequestQueue.open(name=queue_name), **kwargs)

    @crawler.router.default_handler
    async def handler(context) -> None:
        visited.append(context.request.url.replace(BASE, ""))

    stats = (await crawler.run(URLS)).to_dict()
    print("%-28s 进 handler: %-34s 站点收到: %s"
          % (label, sorted(visited), sorted(RECORDS)))
    print("%-28s finished=%d failed=%d total=%d"
          % ("", stats["requests_finished"], stats["requests_failed"], stats["requests_total"]))


async def main() -> None:
    await probe("默认 respect_robots_txt_file=False", "rb-off")
    await probe("respect_robots_txt_file=True", "rb-on", respect_robots_txt_file=True)


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
默认 respect_robots_txt_file=False 进 handler: ['/private/one', '/private/secret', '/public'] 站点收到: ['/private/one', '/private/secret', '/public']
                             finished=3 failed=0 total=3
respect_robots_txt_file=True 进 handler: ['/public']                        站点收到: ['/public', '/robots.txt']
                             finished=1 failed=0 total=1
The `respect_robots_txt_file` option is enabled, but the crawler is not using `ThrottlingRequestManager`. Crawl-delay directives from robots.txt will not be enforced. To enable crawl-delay support, configure the crawler to use `ThrottlingRequestManager` as the request manager.
Some requests were skipped because they were disallowed based on the robots.txt file
  • 打开之后，Disallow: /private/ 下的两条根本没发到站点（靶站只收到 /public 和 /robots.txt 两次请求），也不是「请求后被丢弃」——是在入队前就被 filter_url 拦掉，统计里连 failed 都不算。
  • 想按 robots 抓但只对自己生效的目录，用 context.use_state 或自己维护黑名单；框架这层只提供 robots 一个开关。
  • 礼貌抓取三件套：max_tasks_per_minute 限流、max_requests_per_crawl 限量、respect_robots_txt_file=True 守规矩。

浏览器渲染那一档

数据是 JS 生成到 DOM 里的，HttpCrawler 就只能拿到空壳。同一个 URL 两个爬虫各跑一次（spa1 是前端渲染靶站）：

playwright_vs_http.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# -*- coding: utf-8 -*-
"""同一个前端渲染页面：PlaywrightCrawler 与 HttpCrawler 的差别有多大。"""

import asyncio
import re

import logging
from crawlee.crawlers import HttpCrawler, PlaywrightCrawler
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

URL = "https://spa1.scrape.center/"


async def with_http() -> None:
    crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=1,
                          request_manager=await RequestQueue.open(name="cmp-http"))

    @crawler.router.default_handler
    async def handler(context) -> None:
        html = (await context.http_response.read()).decode("utf-8", errors="replace")
        print("[HttpCrawler] 原始 HTML 长度 = %d，class=\"el-card 出现 %d 次"
              % (len(html), len(re.findall(r'class="el-card', html))))
        print("[HttpCrawler] 正文开头: %s" % html[:96].replace("\n", " ").strip())

    await crawler.run([URL])


async def with_playwright() -> None:
    crawler = PlaywrightCrawler(configure_logging=False, headless=True, max_requests_per_crawl=1,
                                request_manager=await RequestQueue.open(name="cmp-pw"))

    @crawler.router.default_handler
    async def handler(context) -> None:
        await context.page.wait_for_selector("div.el-card", timeout=30000)
        cards = await context.page.locator("div.el-card").count()
        first = await context.page.locator("div.el-card .p-h h2").first.inner_text()
        html = await context.page.content()
        print("[PlaywrightCrawler] 渲染后 div.el-card = %d 张，第一条 = %s" % (cards, first))
        print("[PlaywrightCrawler] 渲染后的 DOM 里 class=\"el-card 出现 %d 次"
              % len(re.findall(r'class="el-card', html)))

    await crawler.run([URL])


async def main() -> None:
    await with_http()
    await with_playwright()


asyncio.run(main())
1
2
3
4
[HttpCrawler] 原始 HTML 长度 = 952，class="el-card 出现 0 次
[HttpCrawler] 正文开头: <!DOCTYPE html><html lang=en><head><meta charset=utf-8><meta http-equiv=X-UA-Compatible content=
[PlaywrightCrawler] 渲染后 div.el-card = 10 张，第一条 = 霸王别姬 - Farewell My Concubine
[PlaywrightCrawler] 渲染后的 DOM 里 class="el-card 出现 20 次
  • 原始 HTML 里 0 张卡片，PlaywrightCrawler 渲染完才有 10 张——差别不在 crawlee，在于有没有真的跑 JS。
  • handler 里必须自己等：await context.page.wait_for_selector("div.el-card")。crawlee 只保证导航完成（goto 返回），不保证 Vue 把数据填上。
  • 两个爬虫在同一个进程里跑时，request_manager 要各给一个具名队列，否则第二个会因为「URL 已经处理过」直接跳过（见 RequestQueue 那节）。

PlaywrightCrawler 的构造参数、以及文档里那些老写法在本机 1.11.1 的下场：

playwright_params.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
# -*- coding: utf-8 -*-
"""PlaywrightCrawler 的真实参数：签名、被删掉的老参数、viewport/locale、屏蔽请求、快照。"""

import asyncio
import inspect

import logging
from crawlee.crawlers import PlaywrightCrawler
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

PORT = 8034
BASE = "http://127.0.0.1:%d" % PORT

PAGE = ("""<html><body>%s</body><script>document.title='ok'</script></html>"""
        % "".join('<img src="/img/%d.png">' % i for i in range(1, 6))).encode()


def start_fixture():
    import threading
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

    seen = []

    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            if self.path.startswith("/img/"):
                seen.append(self.path)
                self.send_response(200)
                self.send_header("Content-Type", "image/png")
                self.send_header("Content-Length", "8")
                self.end_headers()
                self.wfile.write(b"\x89PNG\r\n\x1a\n")
            else:
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(PAGE)))
                self.end_headers()
                self.wfile.write(PAGE)

        def log_message(self, *args):
            pass

    server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server, seen


server, image_hits = start_fixture()

print("PlaywrightCrawler.__init__ 的真实参数（逐个一行，* 之后都是关键字参数）：")
for name, param in inspect.signature(PlaywrightCrawler.__init__).parameters.items():
    if name in ("self", "kwargs"):
        continue
    default = param.default if param.default is not inspect._empty else "必填"
    print("  %-32s = %s" % (name, str(default)[:120]))

print("\n文档里常见、但 1.11.1 已经不存在的参数（逐个传，照抄报错）：")
for name in ["viewport", "launch_context", "launch_options", "block_request_by_url",
             "save_cookies_into_context", "respect_cookies_from_context",
             "max_concurrent_crawling"]:
    try:
        PlaywrightCrawler(**{name: 1})
        print("  %-32s 竟然接受" % name)
    except TypeError as exc:
        print("  %-32s TypeError: %s" % (name, exc))


async def viewport_and_locale() -> None:
    crawler = PlaywrightCrawler(
        configure_logging=False, headless=True, max_requests_per_crawl=1,
        fingerprint_generator=None,
        browser_new_context_options={"viewport": {"width": 1024, "height": 768},
                                     "user_agent": "note-ua-007", "locale": "en-US"},
        request_manager=await RequestQueue.open(name="pw-context"))

    @crawler.router.default_handler
    async def handler(context) -> None:
        print("\nbrowser_new_context_options 生效情况（fingerprint_generator=None）：")
        print("  viewport      =", await context.page.evaluate(
            "window.innerWidth + 'x' + window.innerHeight"))
        print("  user_agent    =", await context.page.evaluate("navigator.userAgent"))
        print("  locale        =", await context.page.evaluate("navigator.language"))

    await crawler.run([BASE + "/"])


async def fingerprint_overrides_ua() -> None:
    crawler = PlaywrightCrawler(
        configure_logging=False, headless=True, max_requests_per_crawl=1,
        browser_new_context_options={"user_agent": "note-ua-007"},
        request_manager=await RequestQueue.open(name="pw-fp"))

    @crawler.router.default_handler
    async def handler(context) -> None:
        ua = await context.page.evaluate("navigator.userAgent")
        print("\nfingerprint_generator 用默认值时，自定义 UA 有没有活下来：",
              "note-ua-007" in ua)

    await crawler.run([BASE + "/"])


async def block_images(with_hook: bool, queue_name: str) -> None:
    image_hits.clear()
    crawler = PlaywrightCrawler(configure_logging=False, headless=True, max_requests_per_crawl=1,
                                request_manager=await RequestQueue.open(name=queue_name))

    if with_hook:
        @crawler.pre_navigation_hook
        async def hook(context) -> None:
            await context.block_requests(url_patterns=[".png"])

    @crawler.router.default_handler
    async def handler(context) -> None:
        imgs = await context.page.locator("img").count()
        print("%s 时页面有 %d 个 <img>，靶站真的收到 %d 个图片请求"
              % ("屏蔽 .png" if with_hook else "不屏蔽", imgs, len(image_hits)))
        image_hits.clear()

    await crawler.run([BASE + "/"])


async def snapshot() -> None:
    crawler = PlaywrightCrawler(configure_logging=False, headless=True, max_requests_per_crawl=1,
                                request_manager=await RequestQueue.open(name="pw-snap"))

    @crawler.router.default_handler
    async def handler(context) -> None:
        snap = await context.get_snapshot()
        print("\nget_snapshot() 返回 %s，公开属性 %s"
              % (type(snap).__name__, [a for a in dir(snap) if not a.startswith("_")]))
        print("  screenshot 类型 %s、非空 %s，html 里有没有 <img>：%s"
              % (type(snap.screenshot).__name__, bool(snap.screenshot), "<img" in snap.html))
        try:
            await context.get_snapshot(screenshot=True, html=True)
        except TypeError as exc:
            print("  照 JS 版文档传参 ->  TypeError: %s" % exc)

    await crawler.run([BASE + "/"])


async def main() -> None:
    await viewport_and_locale()
    await fingerprint_overrides_ua()
    await block_images(False, "pw-no-block")
    await block_images(True, "pw-block")
    await snapshot()


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
PlaywrightCrawler.__init__ 的真实参数（逐个一行，* 之后都是关键字参数）：
  browser_pool                     = None
  browser_type                     = None
  user_data_dir                    = None
  browser_launch_options           = None
  browser_new_context_options      = None
  goto_options                     = None
  fingerprint_generator            = default
  headless                         = None
  use_incognito_pages              = None
  navigation_timeout               = None
文档里常见、但 1.11.1 已经不存在的参数（逐个传，照抄报错）：
  viewport                         TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'viewport'
  launch_context                   TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'launch_context'
  launch_options                   TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'launch_options'
  block_request_by_url             TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'block_request_by_url'
  save_cookies_into_context        TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'save_cookies_into_context'
  respect_cookies_from_context     TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'respect_cookies_from_context'
  max_concurrent_crawling          TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'max_concurrent_crawling'
browser_new_context_options 生效情况（fingerprint_generator=None）：
  viewport      = 1024x768
  user_agent    = note-ua-007
  locale        = en-US
fingerprint_generator 用默认值时，自定义 UA 有没有活下来： False
不屏蔽 时页面有 5 个 <img>，靶站真的收到 5 个图片请求
屏蔽 .png 时页面有 5 个 <img>，靶站真的收到 0 个图片请求
get_snapshot() 返回 PageSnapshot，公开属性 ['html', 'screenshot']
  screenshot 类型 bytes、非空 True，html 里有没有 <img>：True
  照 JS 版文档传参 ->  TypeError: PlaywrightPreNavCrawlingContext.get_snapshot() got an unexpected keyword argument 'screenshot'
  • 真实参数只有 10 个，视口、UA、locale 都塞在 browser_new_context_options 里（Playwright 的 new_context 参数原样透传）。
  • viewport、launch_options、save_cookies_into_context、max_concurrent_crawling 这些文档站上的名字在 1.11.1 直接 TypeError——那批示例是 JS 版或旧版的，别照抄。
  • 屏蔽静态资源用 @crawler.pre_navigation_hook 里的 context.block_requests(url_patterns=[".png"])：靶站一次都没收到图片请求。不给 patterns 时用它的默认黑名单，里面连 .css 都拦，样式敏感的单页应用可能因此渲染不出来。
  • fingerprint_generator 默认是 'default'，它会覆盖你在 browser_new_context_options 里写的 user_agent；想用自己的 UA 就把它设成 None（上面第一段就是这么跑的）。
  • get_snapshot() 在本机版本不接受任何参数，返回 PageSnapshot(html=..., screenshot=...)；文档里 get_snapshot({ screenshot: true }) 的写法会 TypeError。

浏览器侧要并发，改的是 concurrency_settings（max_concurrent_crawling 已经不存在）：

1
2
3
4
5
6
7
8
from crawlee._types import ConcurrencySettings

crawler = PlaywrightCrawler(
    headless=True,
    # 4 个浏览器页面同时跑；不传则由 autoscaling 从 1 往上加
    concurrency_settings=ConcurrencySettings(min_concurrency=4, desired_concurrency=4,
                                             max_concurrency=4),
)

代价很直接：4 并发就是 4 份页面内存，伸缩那条链路上限是 100（并发那节有实测），浏览器爬虫要自己把 max_concurrency 压到机器扛得住的数。

SPA 怎么拿全量

分页控件长什么样，决定了 enqueue_links 有没有用。spa1 的分页是 <li>，没有 href；而它的列表接口能直接按 offset 翻：

spa_pagination.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
# -*- coding: utf-8 -*-
"""前端渲染的站怎么拿全量：浏览器翻页 vs 直接打接口。"""

import asyncio
import json

import logging
from crawlee import Request
from crawlee.crawlers import HttpCrawler, PlaywrightCrawler
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

SITE = "https://spa1.scrape.center"
PAGE = 10
API = SITE + "/api/movie?limit=%d&offset=%%d" % PAGE


async def by_browser() -> None:
    crawler = PlaywrightCrawler(configure_logging=False, headless=True, max_requests_per_crawl=2,
                                request_manager=await RequestQueue.open(name="pg-browser"))
    pages = []
    pagers = []

    @crawler.router.default_handler
    async def handler(context) -> None:
        await context.page.wait_for_selector("div.el-card", timeout=30000)
        got = await context.page.locator("div.el-card .p-h h2").all_inner_texts()
        no = context.request.url.replace(SITE, "") or "/"
        pages.append("%s: %d 条，第一条「%s」" % (no, len(got), got[0]))
        pagers.append("<a> %d 个、<li> %d 个"
                      % (await context.page.locator(".el-pager li a").count(),
                         await context.page.locator(".el-pager li").count()))

    await crawler.run([SITE + "/", SITE + "/page/2"])
    print("浏览器侧:", " | ".join(sorted(pages)))
    print("分页控件:", " | ".join(sorted(set(pagers))), "-> 全是 <li>，没有 href，enqueue_links 发现不了下一页")


async def by_api() -> None:
    crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=40,
                          request_manager=await RequestQueue.open(name="pg-api"))
    names = []
    offsets = []
    page0 = []
    count = []

    @crawler.router.default_handler
    async def handler(context) -> None:
        body = json.loads((await context.http_response.read()).decode("utf-8"))
        names.extend(item["name"] for item in body["results"])
        offset = int(context.request.url.split("offset=")[-1])
        offsets.append(offset)
        count.append(body["count"])
        if offset == 0:
            page0.extend(item["name"] for item in body["results"])
        if offset + PAGE < body["count"]:
            await context.add_requests([Request.from_url(API % (offset + PAGE))])

    await crawler.run([Request.from_url(API % 0)])
    print("接口侧: 接口给的 count=%d，共请求 %d 个 offset（%d..%d），拿到 %d 条（去重后 %d）"
          % (count[0], len(offsets), min(offsets), max(offsets), len(names), len(set(names))))
    print("offset=0 那页的前 3 条:", page0[:3])


async def main() -> None:
    await by_browser()
    await by_api()


asyncio.run(main())
1
2
3
4
浏览器侧: /: 10 条，第一条「霸王别姬 - Farewell My Concubine」 | /page/2: 10 条，第一条「V字仇杀队 - V for Vendetta」
分页控件: <a> 0 个、<li> 8 个 -> 全是 <li>，没有 href，enqueue_links 发现不了下一页
接口侧: 接口给的 count=104，共请求 11 个 offset（0..100），拿到 104 条（去重后 104）
offset=0 那页的前 3 条: ['霸王别姬', '这个杀手不太冷', '肖申克的救赎']
  • 浏览器侧：/ 和 /page/2 都能渲染出 10 条，说明可以直接手工下发 /page/N；但分页控件里 <a> 是 0 个，enqueue_links 一条也发现不了。
  • 接口侧：先打一次 /api/movie?limit=10&offset=0 问出 count=104，再在 handler 里用 context.add_requests 递推下一页的 offset，11 个请求拿满 104 条。
  • 结论：能走接口就别开浏览器。这一站用 HttpCrawler 打接口是 11 个请求、几百字节 JSON；用 PlaywrightCrawler 是 2 个完整页面外加渲染。
  • 接口返回的字段名以实际响应为准（本机这个站是 count / results，不是 items），写代码前先看一眼真实 JSON。

自适应爬虫

AdaptivePlaywrightCrawler 想解决的就是上一条那个抉择：静态能拿到就别开浏览器。它的选择逻辑在 rendering_type_predictor 里，可以换成你自己的：

adaptive_predictor.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
# -*- coding: utf-8 -*-
"""自适应爬虫：同一个 handler，静态解析和浏览器渲染谁来做，由 predictor 决定。"""

import asyncio
import inspect

import logging
from crawlee import Request
from crawlee.crawlers import (AdaptivePlaywrightCrawler, RenderingType, RenderingTypePrediction,
                              RenderingTypePredictor)
from crawlee.crawlers._adaptive_playwright._rendering_type_predictor import (
    DefaultRenderingTypePredictor)
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

SSR = "https://ssr1.scrape.center/page/1"
SPA = "https://spa1.scrape.center/"

print("AdaptivePlaywrightCrawler.__init__ 的参数（逐个一行）：")
for name, param in inspect.signature(AdaptivePlaywrightCrawler.__init__).parameters.items():
    if name == "self":
        continue
    default = param.default if param.default is not inspect._empty else "必填"
    print("  %-28s = %s" % (name, str(default)[:120]))

print("\n工厂方法与默认 predictor：")
print("  with_parsel_static_parser / with_beautifulsoup_static_parser")
print("  DefaultRenderingTypePredictor.__init__%s" % inspect.signature(
    DefaultRenderingTypePredictor.__init__))
print("  RenderingType = %s" % (RenderingType,))


class FixedPredictor(RenderingTypePredictor):
    """手写一个「说什么就是什么」的 predictor，方便看清两条路径的差别。"""

    def __init__(self, kind: str) -> None:
        super().__init__()
        self.kind = kind

    def predict(self, request: Request) -> RenderingTypePrediction:
        return RenderingTypePrediction(rendering_type=self.kind,
                                       detection_probability_recommendation=0.0)

    def store_result(self, request: Request, rendering_type: str) -> None:
        pass


async def run(predictor: RenderingTypePredictor, urls: list[str], queue_name: str) -> list[str]:
    crawler = AdaptivePlaywrightCrawler.with_parsel_static_parser(
        configure_logging=False,
        rendering_type_predictor=predictor,
        playwright_crawler_specific_kwargs={"headless": True},
        request_manager=await RequestQueue.open(name=queue_name),
    )
    log = []

    @crawler.router.default_handler
    async def handler(context) -> None:
        used = "browser" if context._page is not None else "static"
        cards = len(context.parsed_content.css("div.el-card"))
        host = context.request.url.split("/")[2]
        if context._page is None:
            log.append("%s 走 %s，parsed_content 里 %d 张卡" % (host, used, cards))
        else:
            await context.page.wait_for_selector("div.el-card", timeout=30000)
            live = await context.page.locator("div.el-card").count()
            log.append("%s 走 %s，parsed_content 里 %d 张卡，等渲染后 page 里 %d 张"
                       % (host, used, cards, live))

    await crawler.run(urls)
    return sorted(set(log))


async def main() -> None:
    print("\n1) 强制走静态解析（连 SPA 也只发一次 HTTP 请求）：")
    for line in await run(FixedPredictor("static"), [SSR, SPA], "ad-static"):
        print("  " + line)

    print("\n2) 强制走浏览器渲染：")
    for line in await run(FixedPredictor("client only"), [SSR, SPA], "ad-browser"):
        print("  " + line)

    print("\n3) 官方默认的 DefaultRenderingTypePredictor（detection_ratio=0.1，边爬边学）：")
    for line in await run(DefaultRenderingTypePredictor(), [SSR], "ad-auto-ssr"):
        print("  " + line)
    for line in await run(DefaultRenderingTypePredictor(), [SPA], "ad-auto-spa"):
        print("  " + line)


asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
AdaptivePlaywrightCrawler.__init__ 的参数（逐个一行）：
  static_parser                = 必填
  rendering_type_predictor     = None
  result_checker               = None
  result_comparator            = None
  playwright_crawler_specific_kwargs = None
  statistics                   = None
  kwargs                       = 必填
工厂方法与默认 predictor：
  with_parsel_static_parser / with_beautifulsoup_static_parser
  DefaultRenderingTypePredictor.__init__(self, detection_ratio: 'float' = 0.1, *, persistence_enabled: 'bool' = False, persist_state_key: 'str' = 'rendering-type-predictor-state') -> 'None'
  RenderingType = typing.Literal['static', 'client only']
1) 强制走静态解析（连 SPA 也只发一次 HTTP 请求）：
  spa1.scrape.center 走 static，parsed_content 里 0 张卡
  ssr1.scrape.center 走 static，parsed_content 里 10 张卡
2) 强制走浏览器渲染：
  spa1.scrape.center 走 browser，parsed_content 里 0 张卡，等渲染后 page 里 10 张
  ssr1.scrape.center 走 browser，parsed_content 里 10 张卡，等渲染后 page 里 10 张
3) 官方默认的 DefaultRenderingTypePredictor（detection_ratio=0.1，边爬边学）：
  ssr1.scrape.center 走 browser，parsed_content 里 10 张卡，等渲染后 page 里 10 张
  ssr1.scrape.center 走 static，parsed_content 里 10 张卡
  spa1.scrape.center 走 browser，parsed_content 里 0 张卡，等渲染后 page 里 10 张
  spa1.scrape.center 走 static，parsed_content 里 0 张卡
  • RenderingType 只有 'static' 和 'client only' 两个值；工厂方法 with_parsel_static_parser(...) / with_beautifulsoup_static_parser(...) 是官方推荐入口，直接 __init__ 时 static_parser 是必填的。
  • 用固定 predictor 能看清两条路的差别：静态解析对 ssr1 拿 10 张卡、对 spa1 拿 0 张；走浏览器时 ssr1 也是 10 张，而 spa1 的 parsed_content 仍是 0 张——自适应的浏览器分支不会替你等渲染，得在 handler 里自己 wait_for_selector。
  • 默认的 DefaultRenderingTypePredictor(detection_ratio=0.1) 是逻辑回归 + 随机探测，所以同一个 URL 在探测期间 handler 会被跑两遍（browser 一遍、static 一遍），上面第 3 段就是两行。写副作用（入库、发消息）时要按「至少一次」来想。
  • 官方文档说的「先 HTTP 不行才上浏览器」不准确：实际是预测器给概率、按 detection_ratio 随机抽样本再学习。想彻底可控，就自己实现 RenderingTypePredictor 的两个方法。

下载文件的爬虫

FileDownloadCrawler 是专门存二进制的一个类，stream 参数决定整包缓冲还是逐块写：

file_download.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
# -*- coding: utf-8 -*-
"""FileDownloadCrawler：整包缓冲与流式落盘，产物是否一致。"""

import asyncio
import hashlib
import random
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import logging
from crawlee.crawlers import FileDownloadCrawler
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

PORT = 8035
CHUNK = 64 * 1024
BLOB = bytes(random.Random(20261108).getrandbits(8) for _ in range(CHUNK * 32))  # 2 MiB，固定种子


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "application/octet-stream")
        self.send_header("Content-Length", str(len(BLOB)))
        self.end_headers()
        for i in range(0, len(BLOB), CHUNK):
            self.wfile.write(BLOB[i:i + CHUNK])

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
URL = "http://127.0.0.1:%d/big.bin" % PORT

print("靶站上的文件: %d 字节，sha256 前 12 位 %s" % (len(BLOB), hashlib.sha256(BLOB).hexdigest()[:12]))


async def download(stream: bool, queue_name: str, dest: Path) -> None:
    crawler = FileDownloadCrawler(configure_logging=False, stream=stream,
                                  request_manager=await RequestQueue.open(name=queue_name))

    @crawler.router.default_handler
    async def handler(context) -> None:
        with dest.open("wb") as f:
            if stream:
                async for piece in context.http_response.read_stream():
                    f.write(piece)
            else:
                f.write(await context.http_response.read())
        data = dest.read_bytes()
        snap = await context.get_snapshot()
        print("  %s 模式: 存了 %d 字节，sha256 前 12 位 %s，get_snapshot 给的 html=%r、screenshot=%r"
              % ("stream" if stream else "buffer", len(data),
                 hashlib.sha256(data).hexdigest()[:12], snap.html, snap.screenshot))

    await crawler.run([URL])


async def main() -> None:
    await download(False, "fd-buffer", Path("big_buffer.bin"))
    await download(True, "fd-stream", Path("big_stream.bin"))


asyncio.run(main())
server.shutdown()
1
2
3
靶站上的文件: 2097152 字节，sha256 前 12 位 4c7511b61ebd
  buffer 模式: 存了 2097152 字节，sha256 前 12 位 4c7511b61ebd，get_snapshot 给的 html=None、screenshot=None
  stream 模式: 存了 2097152 字节，sha256 前 12 位 4c7511b61ebd，get_snapshot 给的 html=None、screenshot=None
  • 两种模式产物完全一致（同一个 sha256），差别只在内存：stream=False 时整份文件先进内存再落盘，大文件就吃 stream=True。
  • stream=True 时 handler 必须消费 read_stream()，不消费就等于跳过下载。
  • 这个上下文的 get_snapshot() 给的 html/screenshot 都是 None：二进制流不可重读，快照对它没意义。

AI 抽取的两个类

crawlee 1.11.1 里还有两个 experimental 的爬虫类，本机装了依赖但没配模型密钥，所以只测构造与接口，不测抓取效果：

ai_crawlers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
# -*- coding: utf-8 -*-
"""两个「AI 帮忙抽取」的爬虫类：参数与不配密钥时的下场（不发任何外呼请求）。"""

import inspect
import warnings

from stagehand._custom.session import Session as StagehandSession

import logging
from crawlee.crawlers import (PydanticAiCrawler, PydanticAiCrawlingContext, StagehandCrawler,
                              StagehandCrawlingContext)

logging.getLogger().setLevel(logging.CRITICAL)

for cls in (PydanticAiCrawler, StagehandCrawler):
    print("== %s  继承: %s" % (cls.__name__, " <- ".join(b.__name__ for b in cls.__mro__[1:-1])))
    print("   模块:", cls.__module__)
    for name, param in inspect.signature(cls.__init__).parameters.items():
        if name in ("self", "kwargs"):
            continue
        default = param.default if param.default is not inspect._empty else "必填"
        print("   %-34s = %s" % (name, str(default)[:110]))

print("\n没有 OPENAI_API_KEY 时，构造 PydanticAiCrawler 会发生什么：")
with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    try:
        PydanticAiCrawler()
        print("  竟然构造成功了（说明环境里有密钥）")
    except Exception as exc:
        print("  %s: %s" % (type(exc).__name__, str(exc).splitlines()[0][:150]))
    for w in caught:
        print("  %s: %s" % (w.category.__name__, w.message))

print("\n两个类的上下文对象各自多出来的东西：")
from crawlee.crawlers import PydanticAiCrawlingContext, StagehandCrawlingContext

for ctx in (PydanticAiCrawlingContext, StagehandCrawlingContext):
    extra = sorted(a for a in dir(ctx) if not a.startswith("_"))
    print("  %-28s %s" % (ctx.__name__, extra))

print()
print("stagehand SDK 的会话对象（context.page 背后那层）：")
print("  动作方法:", sorted(m for m in dir(StagehandSession) if m in ("act", "extract", "observe")))
print("  act 的参数名:", sorted(inspect.signature(StagehandSession.act).parameters))
print("  extract 的参数名:", sorted(inspect.signature(StagehandSession.extract).parameters))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
== PydanticAiCrawler  继承: AbstractHttpCrawler <- BasicCrawler <- ABC <- Generic
   模块: crawlee.crawlers._pydantic_ai._pydantic_ai_crawler
   model                              = None
   extractor                          = None
== StagehandCrawler  继承: PlaywrightCrawler <- BasicCrawler <- Generic
   模块: crawlee.crawlers._stagehand._stagehand_crawler
   stagehand_options                  = None
   browser_pool                       = None
   user_data_dir                      = None
   headless                           = None
   browser_launch_options             = None
   browser_new_context_options        = None
   goto_options                       = None
   navigation_timeout                 = None
没有 OPENAI_API_KEY 时，构造 PydanticAiCrawler 会发生什么：
  UserError: Set the `OPENAI_API_KEY` environment variable or pass it via `OpenAIProvider(api_key=...)` to use the OpenAI provider. To try Pydantic AI without an A
  UserWarning: The PydanticAiCrawler is experimental and its public API may change in future releases.
两个类的上下文对象各自多出来的东西：
  PydanticAiCrawlingContext    ['create_modified_copy', 'extract_form_requests', 'from_basic_crawling_context', 'from_http_crawling_context', 'from_parsed_http_crawling_context', 'from_parsel_crawling_context', 'get_snapshot', 'html_to_text', 'selector']
  StagehandCrawlingContext     ['create_modified_copy', 'get_snapshot']
stagehand SDK 的会话对象（context.page 背后那层）：
  动作方法: ['act', 'extract', 'observe']
  act 的参数名: ['extra_body', 'extra_headers', 'extra_query', 'page', 'params', 'self', 'timeout']
  extract 的参数名: ['extra_body', 'extra_headers', 'extra_query', 'page', 'params', 'schema', 'self', 'timeout']
  • PydanticAiCrawler 底层是 Parsel（上下文里同时有 selector 和 AI 的 extract），构造当场就要模型密钥，报 pydantic_ai.exceptions.UserError——它不是「跑到一半才失败」，是建对象就失败。
  • StagehandCrawler 是 PlaywrightCrawler 的子类，多了 stagehand_options；handler 里的 context.page 是 Stagehand 包装过的对象，act/extract/observe 三个方法都在。
  • 两个类的 docstring 都明写 experimental、公开 API 未来会变，学习笔记级别的用法先记住「有这两个东西、要模型密钥、别上生产」就够了。

代理怎么配

代理在 crawlee 里是一个独立对象 ProxyConfiguration，塞给任何爬虫的 proxy_configuration 参数就行。它不在 crawlee 顶层导出里，要从 crawlee.proxy_configuration 导入。先把它的三个入口和取代理的方法看清：

proxy_api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
# -*- coding: utf-8 -*-
"""ProxyConfiguration 的参数、轮转规则、自定义函数与分档代理（纯 API，不发请求）。"""

import asyncio
import dataclasses
import inspect

import crawlee
from crawlee import Request
from crawlee.proxy_configuration import ProxyConfiguration, ProxyInfo

print("ProxyConfiguration 在不在 crawlee 顶层导出里:", "ProxyConfiguration" in crawlee.__all__)
print("真实导入路径:", ProxyConfiguration.__module__)
print("构造参数:", inspect.signature(ProxyConfiguration.__init__))
print("取代理的方法:", inspect.signature(ProxyConfiguration.new_proxy_info))

print("\nProxyInfo 的字段（dataclass）:")
for f in dataclasses.fields(ProxyInfo):
    default = f.default if f.default is not dataclasses.MISSING else "必填"
    print("  %-11s %-16s 默认=%s" % (f.name, f.type, default))


async def round_robin() -> None:
    print("\n1) proxy_urls 轮转（None 代表「这一发不走代理」）：")
    pc = ProxyConfiguration(proxy_urls=["http://p1:pw@127.0.0.1:8036",
                                        "http://p2:pw@127.0.0.1:8036", None])
    for i in range(5):
        info = await pc.new_proxy_info(None, None, None)
        print("  第 %d 次 -> %s" % (i + 1, info.url if info else "不套代理"))


async def same_session() -> None:
    print("\n2) 同一个 session_id 固定用同一个代理：")
    pc = ProxyConfiguration(proxy_urls=["http://a@127.0.0.1:8036", "http://b@127.0.0.1:8036",
                                        "http://c@127.0.0.1:8036"])
    got = [(await pc.new_proxy_info("FIXED", None, None)).url for _ in range(4)]
    print("  4 次取到:", sorted(set(got)), "-> 是否全同一个:", len(set(got)) == 1)
    free = sorted({(await pc.new_proxy_info(None, None, None)).url for _ in range(9)})
    print("  不给 session_id 时 9 次取到:", free)


async def custom_function() -> None:
    print("\n3) new_url_function：可同步、可 async、可接 request，返回 None 表示这条不套代理")

    print("  crawlee 固定按 new_url_function(session_id, request) 两个位置参数调用")

    def sync_fn(session_id: str | None, request: Request | None) -> str:
        return "http://sync-%s@127.0.0.1:8036" % (session_id or "nosess")

    async def per_request(session_id: str | None, request: Request | None) -> str | None:
        return None if request.url.endswith("/skip") else "http://per-req@127.0.0.1:8036"

    print("  同步函数, session_id='S1':", (await ProxyConfiguration(
        new_url_function=sync_fn).new_proxy_info("S1", None, None)).url)
    pc = ProxyConfiguration(new_url_function=per_request)
    for url in ["http://example.com/a", "http://example.com/skip"]:
        info = await pc.new_proxy_info(None, Request.from_url(url), None)
        print("  按 request 选, %-24s -> %s" % (url, info.url if info else "不套代理"))
    try:
        await ProxyConfiguration(new_url_function=lambda **kw: "not-a-url").new_proxy_info("S", None, None)
    except Exception as exc:
        print("  返回非法 URL -> %s: %s" % (type(exc).__name__, exc))

    def boom(session_id: str | None, request: Request | None) -> str:
        raise KeyError("我自己的 bug")

    try:
        await ProxyConfiguration(new_url_function=boom).new_proxy_info("S", None, None)
    except Exception as exc:
        print("  函数自己抛异常 -> %s: %s" % (type(exc).__name__, exc))


async def tiers() -> None:
    print("\n4) tiered_proxy_urls：低档失败自动升档（靠 request.last_proxy_tier 记账）")
    pc = ProxyConfiguration(tiered_proxy_urls=[["http://t0@127.0.0.1:8036"],
                                               ["http://t1@127.0.0.1:8036"]])
    req = Request.from_url("http://blocked.example/page")
    for i in range(3):
        info = await pc.new_proxy_info(None, req, None)
        print("  第 %d 次取代理 -> %s (proxy_tier=%s)" % (i + 1, info.url, info.proxy_tier))


async def main() -> None:
    await round_robin()
    await same_session()
    await custom_function()
    await tiers()


asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
ProxyConfiguration 在不在 crawlee 顶层导出里: False
真实导入路径: crawlee.proxy_configuration
构造参数: (self, *, proxy_urls: 'list[str | None] | None' = None, new_url_function: '_NewUrlFunction | None' = None, tiered_proxy_urls: 'list[list[str | None]] | None' = None) -> 'None'
取代理的方法: (self, session_id: 'str | None', request: 'Request | None', proxy_tier: 'int | None') -> 'ProxyInfo | None'
ProxyInfo 的字段（dataclass）:
  url         str              默认=必填
  scheme      str              默认=必填
  hostname    str              默认=必填
  port        int              默认=必填
  username    str              默认=
  password    str              默认=
  session_id  str | None       默认=None
  proxy_tier  int | None       默认=None
1) proxy_urls 轮转（None 代表「这一发不走代理」）：
  第 1 次 -> http://p1:pw@127.0.0.1:8036
  第 2 次 -> http://p2:pw@127.0.0.1:8036
  第 3 次 -> 不套代理
  第 4 次 -> http://p1:pw@127.0.0.1:8036
  第 5 次 -> http://p2:pw@127.0.0.1:8036
2) 同一个 session_id 固定用同一个代理：
  4 次取到: ['http://a@127.0.0.1:8036'] -> 是否全同一个: True
  不给 session_id 时 9 次取到: ['http://a@127.0.0.1:8036', 'http://b@127.0.0.1:8036', 'http://c@127.0.0.1:8036']
3) new_url_function：可同步、可 async、可接 request，返回 None 表示这条不套代理
  crawlee 固定按 new_url_function(session_id, request) 两个位置参数调用
  同步函数, session_id='S1': http://sync-S1@127.0.0.1:8036
  按 request 选, http://example.com/a     -> http://per-req@127.0.0.1:8036
  按 request 选, http://example.com/skip  -> 不套代理
  返回非法 URL -> ValueError: The provided "new_url_function" did not return a valid URL
  函数自己抛异常 -> ValueError: The provided "new_url_function" did not return a valid URL
4) tiered_proxy_urls：低档失败自动升档（靠 request.last_proxy_tier 记账）
  第 1 次取代理 -> http://t0@127.0.0.1:8036 (proxy_tier=0)
  第 2 次取代理 -> http://t1@127.0.0.1:8036 (proxy_tier=1)
  第 3 次取代理 -> http://t0@127.0.0.1:8036 (proxy_tier=0)
  • 三个参数 proxy_urls / tiered_proxy_urls / new_url_function 三选一（docstring 原话），都给会当场报错。
  • proxy_urls 是 round-robin，列表里放一个 None 就表示「这一发直连」，第 3 次取到的就是 None。
  • 同一个 session_id 会固定命中同一个代理——这就是「一个会话绑一个出口 IP」的实现方式；不给 session_id 才会在池子里乱转。
  • new_url_function 是位置参数 (session_id, request) 调用，写错签名不会给你 TypeError 而是被包成 ValueError: The provided "new_url_function" did not return a valid URL——函数内部自己抛的异常也会被这句话盖掉，排查时先单独跑一下这个函数。
  • 分档代理（tiered_proxy_urls）按域名记账：低档在这个域名上记过错，下一次就升档；ProxyInfo.proxy_tier 告诉你当前是第几档。

光看 API 不够，本机起了一个转发代理和一个回显源站，真走一遍：

proxy_live.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
# -*- coding: utf-8 -*-
"""真走一遍代理：本机起一个转发代理 + 一个回显源站，看代理收到的、源站收到的分别是什么。"""

import asyncio
import base64
import json
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import logging
from crawlee import Request
from crawlee.crawlers import HttpCrawler
from crawlee.proxy_configuration import ProxyConfiguration
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

PROXY_PORT, ORIGIN_PORT = 8036, 8037
ORIGIN = "http://127.0.0.1:%d/echo" % ORIGIN_PORT


class OriginHandler(BaseHTTPRequestHandler):
    """源站：把收到的头原样回出来（代理加的 Via / X-Forwarded-For 也在里面）。"""

    def do_GET(self):
        body = json.dumps({k.lower(): v for k, v in self.headers.items()}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


class ProxyHandler(BaseHTTPRequestHandler):
    """转发代理：用户名以 dead 开头就假装这个代理被封，否则带上 Via/XFF 转发给源站。"""

    def do_GET(self):
        auth = self.headers.get("Proxy-Authorization", "")
        user = ""
        if auth.startswith("Basic "):
            user = base64.b64decode(auth[6:]).decode().split(":")[0]
        if user.startswith("dead"):
            self.send_response(403)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        forwarded = urllib.request.Request(self.path, headers={"Via": "1.1 local-proxy",
                                                               "X-Forwarded-For": "203.0.113.9",
                                                               "X-Proxy-User": user})
        with urllib.request.urlopen(forwarded, timeout=10) as resp:
            payload = resp.read()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, *args):
        pass


origin = ThreadingHTTPServer(("127.0.0.1", ORIGIN_PORT), OriginHandler)
proxy = ThreadingHTTPServer(("127.0.0.1", PROXY_PORT), ProxyHandler)
threading.Thread(target=origin.serve_forever, daemon=True).start()
threading.Thread(target=proxy.serve_forever, daemon=True).start()


async def crawl(proxy_configuration: ProxyConfiguration, urls: list[str], queue_name: str) -> None:
    crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=len(urls),
                          proxy_configuration=proxy_configuration,
                          request_manager=await RequestQueue.open(name=queue_name))
    rows = []

    @crawler.router.default_handler
    async def handler(context) -> None:
        head = json.loads((await context.http_response.read()).decode("utf-8"))
        info = context.proxy_info
        rows.append("%s | 代理=%s tier=%s | 源站看到 Via=%r XFF=%r 经 %s 转的"
                    % (context.request.url.split("?")[-1], info.username, info.proxy_tier,
                       head.get("via"), head.get("x-forwarded-for"), head.get("x-proxy-user")))

    await crawler.run([Request.from_url(u) for u in urls])
    for row in sorted(rows):
        print("  " + row)


async def main() -> None:
    print("1) 两个普通代理轮转（URL 里带序号方便对照）：")
    await crawl(ProxyConfiguration(proxy_urls=["http://p1:pw@127.0.0.1:%d" % PROXY_PORT,
                                               "http://p2:pw@127.0.0.1:%d" % PROXY_PORT]),
                ["%s?n=1" % ORIGIN, "%s?n=2" % ORIGIN], "px-round")

    print("\n2) 不配代理，源站看到的就没有 Via：")
    await crawl(ProxyConfiguration(proxy_urls=["http://127.0.0.1:%d" % ORIGIN_PORT]),
                ["%s?n=3" % ORIGIN], "px-direct")

    print("\n3) 分档代理：tier0 是个坏代理（用户名 dead 开头），自动升到 tier1：")
    await crawl(ProxyConfiguration(tiered_proxy_urls=[
        ["http://dead0:pw@127.0.0.1:%d" % PROXY_PORT],
        ["http://good1:pw@127.0.0.1:%d" % PROXY_PORT]]), ["%s?n=4" % ORIGIN], "px-tier")


asyncio.run(main())
proxy.shutdown()
origin.shutdown()
1
2
3
4
5
6
7
1) 两个普通代理轮转（URL 里带序号方便对照）：
  n=1 | 代理=p1 tier=None | 源站看到 Via='1.1 local-proxy' XFF='203.0.113.9' 经 p1 转的
  n=2 | 代理=p2 tier=None | 源站看到 Via='1.1 local-proxy' XFF='203.0.113.9' 经 p2 转的
2) 不配代理，源站看到的就没有 Via：
  n=3 | 代理= tier=None | 源站看到 Via=None XFF=None 经 None 转的
3) 分档代理：tier0 是个坏代理（用户名 dead 开头），自动升到 tier1：
  n=4 | 代理=good1 tier=1 | 源站看到 Via='1.1 local-proxy' XFF='203.0.113.9' 经 good1 转的
  • 源站看到的 Via / X-Forwarded-For 是代理加的，context.proxy_info 拿到的就是这一发实际用的代理。
  • 第二个例子故意不套代理（把源站自己填进 proxy_urls），Via 就是 None——判断「代理到底走没走」，看源站收到的头最靠谱，别只看代码里配了。
  • 第三个例子里 tier0 那个代理专门返回 403，于是 proxy_tier 变成 1、由 tier1 成功：这就是「这批 IP 被站点拉黑就自动换下一批」的机制。
  • 代理 URL 里的密码在 ProxyInfo 的 repr 里被掩成 ***，日志里不会泄出来。

会话与封禁

use_session_pool=True 是默认开的。会话对象负责记账：什么状态码算被封、错误分涨到多少就退役。

session_pool_demo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
# -*- coding: utf-8 -*-
"""会话池：Session 的默认值、好坏打分，以及被 403 之后的自动换号行为。"""

import asyncio
import inspect
import logging
import threading
from datetime import timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee.crawlers import HttpCrawler
from crawlee.sessions import Session, SessionPool
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

PORT = 8038
HITS: dict[str, int] = {}


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        HITS[self.path] = HITS.get(self.path, 0) + 1
        code = 403 if self.path.startswith("/blocked") else (404 if self.path.startswith("/nope") else 200)
        self.send_response(code)
        self.send_header("Content-Type", "text/html")
        self.end_headers()
        self.wfile.write(b"<html><body>ok</body></html>")

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()

print("Session.__init__ 的默认值：")
for name, param in inspect.signature(Session.__init__).parameters.items():
    if name == "self":
        continue
    print("  %-24s = %s" % (name, param.default))

print("\nSession 的方法/属性:", sorted(a for a in dir(Session) if not a.startswith("_")))
print("SessionPool.__init__:", inspect.signature(SessionPool.__init__))

print("\n1) 哪些状态码算「被封」（用公开方法 is_blocked_status_code 逐个试）：")
probe = Session()
codes = [200, 401, 402, 403, 404, 429, 500, 503]
print("  会被判定封禁:", [c for c in codes if probe.is_blocked_status_code(status_code=c)])
print("  带上 ignore_http_error_status_codes={403} 之后:",
      [c for c in codes if probe.is_blocked_status_code(status_code=c,
                                        ignore_http_error_status_codes={403})])

s = Session()
print("\n2) 错误分打分（max_error_score=3.0、error_score_decrement=0.5）：")
for i in range(4):
    s.mark_bad()
    print("  mark_bad 第 %d 次 -> error_score=%.1f is_blocked=%s"
          % (i + 1, s.error_score, s.is_blocked))
s.mark_good()
print("  mark_good 一次 -> error_score=%.1f is_blocked=%s" % (s.error_score, s.is_blocked))

print("\n3) 用量与寿命：")
print("  Session(usage_count=50).is_max_usage_count_reached =",
      Session(usage_count=50).is_max_usage_count_reached)
print("  Session(max_age=timedelta(seconds=-1)).is_expired =",
      Session(max_age=timedelta(seconds=-1)).is_expired)


async def try_one(queue_name: str, path: str, **kwargs) -> None:
    HITS.clear()
    crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=1,
                          request_manager=await RequestQueue.open(name=queue_name), **kwargs)
    entered = []
    failed = []

    @crawler.router.default_handler
    async def handler(context) -> None:
        entered.append(context.http_response.status_code)

    @crawler.failed_request_handler
    async def on_failed(context, error) -> None:
        failed.append("%s: %s" % (type(error).__name__, error))

    await crawler.run(["http://127.0.0.1:%d%s" % (PORT, path)])
    print("  靶站被打 %d 次，handler 正常返回 %s 次，失败原因: %s"
          % (HITS.get(path, 0), len(entered), failed[-1] if failed else "无"))


async def main() -> None:
    print("\n4) 默认碰上 403（判定会话被封，换会话重试）：")
    await try_one("ss-403", "/blocked")

    print("\n5) 同一发 403，但关掉会话池：")
    await try_one("ss-403-nopool", "/blocked", use_session_pool=False)

    print("\n6) 同一发 403，但 max_session_rotations=2：")
    await try_one("ss-403-rot2", "/blocked", max_session_rotations=2)

    print("\n7) 403 当成正常响应：")
    await try_one("ss-403-ignore", "/blocked", ignore_http_error_status_codes=[403])

    print("\n8) 对照：404 默认既不重试也不换会话")
    await try_one("ss-404", "/nope")


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
Session.__init__ 的默认值：
  id                       = None
  max_age                  = 0:50:00
  user_data                = None
  max_error_score          = 3.0
  error_score_decrement    = 0.5
  created_at               = None
  usage_count              = 0
  max_usage_count          = 50
  error_score              = 0.0
  cookies                  = None
  blocked_status_codes     = None
Session 的方法/属性: ['cookies', 'error_score', 'expires_at', 'from_model', 'get_state', 'id', 'is_blocked', 'is_blocked_status_code', 'is_expired', 'is_max_usage_count_reached', 'is_usable', 'mark_bad', 'mark_good', 'retire', 'usage_count', 'user_data']
SessionPool.__init__: (self, *, max_pool_size: 'int' = 1000, create_session_settings: 'dict | None' = None, create_session_function: 'CreateSessionFunctionType | None' = None, event_manager: 'EventManager | None' = None, persistence_enabled: 'bool' = False, persist_state_kvs_name: 'str | None' = None, persist_state_key: 'str' = 'CRAWLEE_SESSION_POOL_STATE') -> 'None'
1) 哪些状态码算「被封」（用公开方法 is_blocked_status_code 逐个试）：
  会被判定封禁: [401, 403, 429]
  带上 ignore_http_error_status_codes={403} 之后: [401, 429]
2) 错误分打分（max_error_score=3.0、error_score_decrement=0.5）：
  mark_bad 第 1 次 -> error_score=1.0 is_blocked=False
  mark_bad 第 2 次 -> error_score=2.0 is_blocked=False
  mark_bad 第 3 次 -> error_score=6.0 is_blocked=True
  mark_bad 第 4 次 -> error_score=10.0 is_blocked=True
  mark_good 一次 -> error_score=12.5 is_blocked=True
3) 用量与寿命：
  Session(usage_count=50).is_max_usage_count_reached = True
  Session(max_age=timedelta(seconds=-1)).is_expired = True
4) 默认碰上 403（判定会话被封，换会话重试）：
  靶站被打 10 次，handler 正常返回 0 次，失败原因: SessionError: Assuming the session is blocked based on HTTP status code 403
5) 同一发 403，但关掉会话池：
  靶站被打 1 次，handler 正常返回 0 次，失败原因: HttpClientStatusCodeError: Client error status code returned (status code: 403).
6) 同一发 403，但 max_session_rotations=2：
  靶站被打 2 次，handler 正常返回 0 次，失败原因: SessionError: Assuming the session is blocked based on HTTP status code 403
7) 403 当成正常响应：
  靶站被打 1 次，handler 正常返回 1 次，失败原因: 无
8) 对照：404 默认既不重试也不换会话
  靶站被打 1 次，handler 正常返回 0 次，失败原因: HttpClientStatusCodeError: Client error status code returned (status code: 404).
  • 默认认为 401 / 403 / 429 是「这个会话被封了」，ignore_http_error_status_codes 能把某个码从这份名单里摘掉。
  • mark_bad 一次加 1 分，超过 max_error_score=3.0 就 is_blocked；一旦不可用，crawlee 还会顺手 retire() 再加 3 分，所以分数是 1 → 2 → 6 → 10 这样跳的，不是每次都 +1。
  • 碰上 403 时默认会换会话重试，一共 max_session_rotations=10 次，最后报 SessionError；关掉会话池就变成一次性的 HttpClientStatusCodeError。这两句话是同一个现象的两种表现，调「要不要换号」时看这个差别。
  • 404 不在重试名单里：HttpClientStatusCodeError 属于 4xx，默认不重试（重试那节有对照）。
  • SessionPool 的 max_pool_size=1000、persist_state_key='CRAWLEE_SESSION_POOL_STATE'，默认 persistence_enabled=False，也就是说会话池的状态默认不入库。

HTTP 客户端与指纹

crawlee 自己不发请求，它包了三个客户端：impit（默认）、httpx、curl_cffi。三个客户端的参数和它们实际发出的头：

clients_and_fingerprint.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
# -*- coding: utf-8 -*-
"""三个 HTTP 客户端的参数与它们实际发出的头，以及 fingerprint_suite 能给什么。"""

import asyncio
import inspect
import json
import logging
import re
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee.crawlers import HttpCrawler
from crawlee.fingerprint_suite import DefaultFingerprintGenerator, HeaderGenerator, HeaderGeneratorOptions
from crawlee.http_clients import CurlImpersonateHttpClient, HttpxHttpClient, ImpitHttpClient
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

PORT = 8039
BASE = "http://127.0.0.1:%d" % PORT


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps({k.lower(): v for k, v in self.headers.items()}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()

for cls in (ImpitHttpClient, HttpxHttpClient, CurlImpersonateHttpClient):
    print("%s 的参数:" % cls.__name__)
    for name, param in inspect.signature(cls.__init__).parameters.items():
        if name in ("self", "args", "kwargs"):
            continue
        shown = type(param.default).__name__ if "object at 0x" in str(param.default) else str(param.default)[:70]
        print("  %-30s = %s" % (name, shown))
    extra = {"async_client_kwargs": "透传给底层库"} if cls is not CurlImpersonateHttpClient else {"async_session_kwargs": "透传给 curl_cffi"}
    print("  还吃 %s" % extra)

print("\ncrawlee 默认用的是哪个客户端：")
crawler_probe = HttpCrawler(configure_logging=False)
print("  HttpCrawler()._http_client =", type(crawler_probe._http_client).__name__)


async def one(label: str, queue_name: str, **kwargs) -> None:
    crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=1,
                          request_manager=await RequestQueue.open(name=queue_name), **kwargs)
    got = {}

    @crawler.router.default_handler
    async def handler(context) -> None:
        got.update(json.loads((await context.http_response.read()).decode("utf-8")))

    await crawler.run([BASE + "/echo"])
    ua = got.get("user-agent", "")
    # HttpxHttpClient 默认的 header_generator 每次随机抽一个浏览器，只有它能打印出抖动
    shown = "UA 由 HeaderGenerator 随机抽" if "HttpxHttpClient" in label and "None" not in label \
        else sorted(set(re.findall(r"\b(Chrome|Firefox|Safari|Edg)/", ua)))
    print("  %-34s 头 %2d 个 %-52s 浏览器=%s"
          % (label, len(got), sorted(got)[:6], shown))


async def fingerprints() -> None:
    print("\nfingerprint_suite（browserforge 1.2.4 在本机能给什么）：")
    hg = HeaderGenerator()
    for kind in ["chrome", "firefox"]:
        head = hg.get_specific_headers(browser_type=kind)
        fam = sorted(set(re.findall(r"\b(Chrome|Firefox|Safari|Edg)/", head["user-agent"])))
        print("  get_specific_headers(browser_type=%-8s) -> %d 个头，UA 家族 %s，有 sec-ch-ua: %s"
              % (repr(kind), len(head), fam, "sec-ch-ua" in head))
    fp = DefaultFingerprintGenerator().generate()
    print("  DefaultFingerprintGenerator().generate() -> %s，headers %d 个，screen 有宽度 %s，mockWebRTC=%s"
          % (type(fp).__name__, len(fp.headers), fp.screen.width > 0, fp.mockWebRTC))
    print("  HeaderGeneratorOptions 的字段:", list(HeaderGeneratorOptions.model_fields))
    print("  注意：HttpCrawler 上没有 use_fingerprints / fingerprint_generator 参数，只有 PlaywrightCrawler 有")
    try:
        HttpCrawler(fingerprint_generator=None)
    except TypeError as exc:
        print("  传给 HttpCrawler ->  TypeError: %s" % exc)


async def main() -> None:
    print("\n1) 各客户端实际发出的头（本机靶站回显）：")
    await one("ImpitHttpClient() 默认", "cl-impit")
    await one("ImpitHttpClient(browser=None)", "cl-impit-none",
              http_client=ImpitHttpClient(browser=None))
    await one("ImpitHttpClient(browser='chrome')", "cl-impit-chrome",
              http_client=ImpitHttpClient(browser="chrome"))
    await one("ImpitHttpClient(http3=True)", "cl-impit-h3",
              http_client=ImpitHttpClient(http3=True))
    await one("HttpxHttpClient()", "cl-httpx", http_client=HttpxHttpClient())
    await one("HttpxHttpClient(header_generator=None)", "cl-httpx-nog",
              http_client=HttpxHttpClient(header_generator=None))
    await one("CurlImpersonateHttpClient()", "cl-curl",
              http_client=CurlImpersonateHttpClient())
    await one("CurlImpersonateHttpClient(impersonate='firefox133')", "cl-curl-ff",
              http_client=CurlImpersonateHttpClient(impersonate="firefox133"))
    await fingerprints()


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
ImpitHttpClient 的参数:
  persist_cookies_per_session    = True
  http3                          = False
  verify                         = True
  browser                        = firefox
  follow_redirects               = True
  max_redirects                  = 20
  async_client_kwargs            = <class 'inspect._empty'>
  还吃 {'async_client_kwargs': '透传给底层库'}
HttpxHttpClient 的参数:
  persist_cookies_per_session    = True
  http1                          = True
  http2                          = True
  verify                         = True
  header_generator               = HeaderGenerator
  async_client_kwargs            = <class 'inspect._empty'>
  还吃 {'async_client_kwargs': '透传给底层库'}
CurlImpersonateHttpClient 的参数:
  persist_cookies_per_session    = True
  async_session_kwargs           = <class 'inspect._empty'>
  还吃 {'async_session_kwargs': '透传给 curl_cffi'}
crawlee 默认用的是哪个客户端：
  HttpCrawler()._http_client = ImpitHttpClient
1) 各客户端实际发出的头（本机靶站回显）：
  ImpitHttpClient() 默认               头 11 个 ['accept', 'accept-encoding', 'accept-language', 'host', 'priority', 'sec-fetch-dest'] 浏览器=['Firefox']
  ImpitHttpClient(browser=None)      头  3 个 ['accept', 'accept-encoding', 'host']                浏览器=[]
  ImpitHttpClient(browser='chrome')  头 13 个 ['accept', 'accept-encoding', 'accept-language', 'host', 'sec-ch-ua', 'sec-ch-ua-mobile'] 浏览器=['Chrome', 'Safari']
  ImpitHttpClient(http3=True)        头 11 个 ['accept', 'accept-encoding', 'accept-language', 'host', 'priority', 'sec-fetch-dest'] 浏览器=['Firefox']
  HttpxHttpClient()                  头  6 个 ['accept', 'accept-encoding', 'accept-language', 'connection', 'host', 'user-agent'] 浏览器=UA 由 HeaderGenerator 随机抽
  HttpxHttpClient(header_generator=None) 头  5 个 ['accept', 'accept-encoding', 'connection', 'host', 'user-agent'] 浏览器=[]
  CurlImpersonateHttpClient()        头 17 个 ['accept', 'accept-encoding', 'accept-language', 'cache-control', 'connection', 'host'] 浏览器=['Chrome', 'Safari']
  CurlImpersonateHttpClient(impersonate='firefox133') 头 15 个 ['accept', 'accept-encoding', 'accept-language', 'connection', 'host', 'http2-settings'] 浏览器=['Firefox']
fingerprint_suite（browserforge 1.2.4 在本机能给什么）：
  get_specific_headers(browser_type='chrome') -> 12 个头，UA 家族 ['Chrome', 'Safari']，有 sec-ch-ua: True
  get_specific_headers(browser_type='firefox') -> 10 个头，UA 家族 ['Firefox']，有 sec-ch-ua: False
  DefaultFingerprintGenerator().generate() -> Fingerprint，headers 12 个，screen 有宽度 True，mockWebRTC=False
  HeaderGeneratorOptions 的字段: ['browsers', 'operating_systems', 'devices', 'locales', 'http_version', 'strict']
  注意：HttpCrawler 上没有 use_fingerprints / fingerprint_generator 参数，只有 PlaywrightCrawler 有
  传给 HttpCrawler ->  TypeError: BasicCrawler.__init__() got an unexpected keyword argument 'fingerprint_generator'
  • 默认客户端是 ImpitHttpClient，它的 browser 默认值就是 'firefox'，所以裸跑也带 11 个 Firefox 的头（第一节的默认请求头是同一个实验的另一种问法）。
  • browser=None 只剩 3 个头（accept、accept-encoding、host），这种「干净得反常」的请求反而最容易被拦。
  • CurlImpersonateHttpClient 是唯一能伪装 TLS/HTTP2 指纹的客户端，impersonate 的名字要写 firefox133 这种（firefox_120 直接 ImpersonateError），名字清单在 curl_cffi.requests.BrowserType 里。
  • HttpxHttpClient 会默认挂一个 HeaderGenerator，所以它自己也能补出浏览器头；header_generator=None 就退回 httpx 原生那 5 个。
  • 指纹这套（DefaultFingerprintGenerator、HeaderGenerator）真正接进爬虫的地方只有 PlaywrightCrawler(fingerprint_generator=...)；HTTP 爬虫没有这个参数，要伪装就只能自己往 Request.headers 里塞生成的头。

uniqueKey 与存储结构

去重的本体是 uniqueKey：crawlee 先对 URL 做归一化（去 fragment、尾部斜杠、查询参数排序），再拿它当主键。这份归一化规则、以及「跑完一次磁盘上到底留下了什么」，都在下面这段里：

unique_key_and_storage.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
# -*- coding: utf-8 -*-
"""uniqueKey 的规则：哪些 URL 会被判成同一个请求；跑完之后 storage 目录长什么样。"""

import asyncio
import shutil
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

from crawlee import Request
from crawlee._request import compute_unique_key
from crawlee.crawlers import ParselCrawler
from crawlee.storages import Dataset, KeyValueStore, RequestQueue

PORT = 8040
BASE = "http://127.0.0.1:%d" % PORT
HITS: list[str] = []


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        HITS.append(self.path)
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.end_headers()
        self.wfile.write(b"<html><body><p>ok</p></body></html>")

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()

print("1) uniqueKey 怎么算（同一个 key 就是同一个请求，会被去重）：")
rows = [
    (BASE + "/one", "GET", None, False),
    (BASE + "/one/", "GET", None, False),
    (BASE + "/two#frag", "GET", None, False),
    (BASE + "/two#frag", "GET", None, True),
    (BASE + "/three?a=2&a=1", "GET", None, False),
    (BASE + "/four", "POST", b'{"a":1}', False),
    (BASE + "/four", "POST", b'{"a":2}', False),
    (BASE + "/four", "POST", b'{"a":2}', True),
    ("HTTPS://127.0.0.1:%d/five" % PORT, "GET", None, False),
]
for url, method, payload, extended in rows:
    kwargs = {"method": method}
    if payload:
        kwargs["payload"] = payload
    if extended:
        kwargs["use_extended_unique_key"] = True
    print("  %-40s %-4s extended=%-5s -> %s"
          % (url, method, extended, Request.from_url(url, **kwargs).unique_key))
print("  compute_unique_key 也能直接吃字符串:", compute_unique_key(BASE + "/three?a=1&a=2"))


async def layout() -> None:
    shutil.rmtree("storage", ignore_errors=True)
    crawler = ParselCrawler(configure_logging=False, max_requests_per_crawl=10,
                            request_manager=await RequestQueue.open(name="uk-layout"))

    @crawler.router.default_handler
    async def handler(context) -> None:
        await context.push_data({"path": context.request.url.replace(BASE, ""),
                                 "text": context.selector.css("p::text").get()})

    await crawler.run([BASE + "/one", BASE + "/one/", BASE + "/two#frag", BASE + "/two",
                       BASE + "/three?a=1&a=2", BASE + "/three?a=2&a=1"])
    print("\n2) 6 条种子 URL 实际请求了几次:", len(HITS), sorted(HITS))

    meta = await (await RequestQueue.open(name="uk-layout")).get_metadata()
    print("  队列: total=%s handled=%s pending=%s"
          % (meta.total_request_count, meta.handled_request_count, meta.pending_request_count))

    items = await (await Dataset.open()).list_items()
    print("  默认数据集条数:", len(items), "path 全貌:", sorted(i["path"] for i in items))

    kvs = await KeyValueStore.open()
    keys = [("__RQ_STATE_<队列 id>" if k.key.startswith("__RQ_STATE_") else k.key)
            for k in await kvs.list_keys()]
    print("  默认 KVS 的 key:", sorted(keys))

    tree = sorted(Path("storage").rglob("*"))
    print("  storage 目录 %d 项（__RQ_STATE_ 后面是本进程随机生成的队列 id，这里统一写成 <队列 id>）:"
          % len(tree))
    for p in tree:
        name = p.relative_to("storage").as_posix()
        if "__RQ_STATE_" in name:
            name = "__RQ_STATE_<队列 id>" + name.split("__RQ_STATE_")[1][20:]
        print("    %s %s" % ("D" if p.is_dir() else "F", name))


asyncio.run(layout())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
1) uniqueKey 怎么算（同一个 key 就是同一个请求，会被去重）：
  http://127.0.0.1:8040/one                GET  extended=False -> http://127.0.0.1:8040/one
  http://127.0.0.1:8040/one/               GET  extended=False -> http://127.0.0.1:8040/one
  http://127.0.0.1:8040/two#frag           GET  extended=False -> http://127.0.0.1:8040/two
  http://127.0.0.1:8040/two#frag           GET  extended=True  -> GET|e3b0c442|e3b0c442|http://127.0.0.1:8040/two
  http://127.0.0.1:8040/three?a=2&a=1      GET  extended=False -> http://127.0.0.1:8040/three?a=1&a=2
  http://127.0.0.1:8040/four               POST extended=False -> http://127.0.0.1:8040/four
  http://127.0.0.1:8040/four               POST extended=False -> http://127.0.0.1:8040/four
  http://127.0.0.1:8040/four               POST extended=True  -> POST|e3b0c442|7e8059f4|http://127.0.0.1:8040/four
  HTTPS://127.0.0.1:8040/five              GET  extended=False -> https://127.0.0.1:8040/five
  compute_unique_key 也能直接吃字符串: http://127.0.0.1:8040/three?a=1&a=2
2) 6 条种子 URL 实际请求了几次: 3 ['/one', '/three?a=1&a=2', '/two']
  队列: total=3 handled=3 pending=0
  默认数据集条数: 3 path 全貌: ['/one', '/three?a=1&a=2', '/two#frag']
  默认 KVS 的 key: ['__CRAWLER_STATISTICS_0', '__RQ_STATE_<队列 id>']
  storage 目录 19 项（__RQ_STATE_ 后面是本进程随机生成的队列 id，这里统一写成 <队列 id>）:
    D datasets
    D datasets/default
    F datasets/default/000000001.json
    F datasets/default/000000002.json
    F datasets/default/000000003.json
    F datasets/default/__metadata__.json
    D key_value_stores
    D key_value_stores/default
    F key_value_stores/default/__CRAWLER_STATISTICS_0
    F key_value_stores/default/__CRAWLER_STATISTICS_0.__metadata__.json
    F key_value_stores/default/__metadata__.json
    F __RQ_STATE_<队列 id>
    F __RQ_STATE_<队列 id>metadata__.json
    D request_queues
    D request_queues/uk-layout
    F request_queues/uk-layout/44ec7a8535b609b.json
    F request_queues/uk-layout/__metadata__.json
    F request_queues/uk-layout/e03e3065d665e58.json
    F request_queues/uk-layout/f9236de9050d6fb.json
  • 归一化会吃掉这些差异：/one 与 /one/、/two 与 /two#frag、?a=2&a=1 与 ?a=1&a=2、大小写与默认端口，POST 的 body 默认不参与去重（同 URL 不同 payload 是同一个请求）。
  • 想按 payload 区分，构造时给 use_extended_unique_key=True，key 变成 POST|<payload 摘要>|<headers 摘要>|<url>。
  • 因为 6 条种子只归并成 3 个请求，靶站确实只被打 3 次，队列里 total=handled=3，数据集 3 条——去重发生在入队时，不是抓完之后丢弃。
  • 目录是三类存储各一个文件夹：datasets/default/000000001.json（序号固定 9 位）、key_value_stores/default/（爬虫统计 __CRAWLER_STATISTICS_0 和队列状态 __RQ_STATE_<队列 id> 都在这）、request_queues/<队列名>/<uniqueKey 摘要>.json。
  • 队列名只允许小写字母、数字和中间的连字符，my_queue 会 ValueError（见报错速查那节）。

配置与环境变量

Configuration 是一个 pydantic BaseSettings，一共 15 个字段，每个字段都能用 CRAWLEE_* 或 APIFY_* 环境变量给：

configuration_and_purge.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
# -*- coding: utf-8 -*-
"""Configuration 的 15 个配置项、环境变量的实际效果，以及 purge_on_start 的边界。"""

import asyncio
import os
import shutil
import subprocess
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee.configuration import Configuration
from crawlee.storages import Dataset

PORT = 8041
BASE = "http://127.0.0.1:%d" % PORT


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.end_headers()
        self.wfile.write(b"<html><body>ok</body></html>")

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()

print("1) Configuration 的全部字段（pydantic BaseSettings，字段名对应大写环境变量）：")
for name, field in Configuration.model_fields.items():
    aliases = getattr(field.validation_alias, "choices", None) or (field.validation_alias,)
    print("  %-26s 默认=%-24s 环境变量=%s"
          % (name, str(field.default)[:24], sorted(a for a in aliases if a)))

print("\n2) 环境变量单独起子进程实测（父进程环境保持干净）：")
CASES = [
    ("CRAWLEE_STORAGE_DIR", "./env_storage", "storage_dir"),
    ("CRAWLEE_LOG_LEVEL", "DEBUG", "log_level"),
    ("CRAWLEE_PURGE_ON_START", "false", "purge_on_start"),
    ("CRAWLEE_HEADLESS", "false", "headless"),
    ("CRAWLEE_MAX_CLIENT_ERRORS", "7", "max_client_errors"),
    ("CRAWLEE_INTERNAL_TIMEOUT", "60", "internal_timeout"),
    ("CRAWLEE_INTERNAL_TIMEOUT", "PT60S", "internal_timeout"),
    ("CRAWLEE_PERSIST_STATE_INTERVAL_MILLIS", "5000", "persist_state_interval"),
    ("CRAWLEE_LOG_LEVEL", "NOPE", "log_level"),
    ("CRAWLEE_DEDUPLICATION_POLICY", "EXTENDED", "log_level"),
    ("CRAWLEE_MAX_CRAWLING_TIME", "60", "log_level"),
    ("CRAWLEE_PROXY_URLS", "http://127.0.0.1:8080", "log_level"),
    ("APIFY_LOCAL_STORAGE_DIR", "./apify_storage", "storage_dir"),
    ("ACTOR_MEMORY_MBYTES", "2048", "memory_mbytes"),
]
SNIPPET = ("import json;"
           "from crawlee.configuration import Configuration;"
           "c = Configuration();"
           "print(json.dumps({'NAME': str(c.NAME)}, ensure_ascii=False))")
for var, value, field in CASES:
    env = {k: v for k, v in os.environ.items()
           if not k.startswith(("CRAWLEE_", "APIFY_", "ACTOR_"))}
    env[var] = value
    proc = subprocess.run([sys.executable, "-X", "utf8", "-c", SNIPPET.replace("NAME", field)],
                          capture_output=True, text=True, env=env, timeout=60)
    got = proc.stdout.strip() or proc.stderr.strip().splitlines()[-1][:110]
    print("  %-40s 读 %-24s -> %s" % ("%s=%s" % (var, value), field, got))

WORKER_TPL = """
import asyncio
from crawlee.crawlers import ParselCrawler
from crawlee.storages import Dataset, RequestQueue

BASE = @@BASE@@


async def run():
    crawler = ParselCrawler(configure_logging=False,
                            request_manager=await RequestQueue.open(name=@@Q@@))

    @crawler.router.default_handler
    async def handler(context):
        await context.push_data({"tag": context.request.url.replace(BASE, "")})

    await crawler.run([BASE + "/a", BASE + "/b"])
    items = await (await Dataset.open()).list_items()
    keep = await (await Dataset.open(name="keepme")).list_items()
    print("默认数据集 %d 条 %s；具名 keepme 还剩 %d 条"
          % (len(items), sorted(i["tag"] for i in items), len(keep)))


asyncio.run(run())
"""


def run_worker(purge: bool, queue_name: str) -> None:
    env = {k: v for k, v in os.environ.items() if not k.startswith(("CRAWLEE_", "APIFY_", "ACTOR_"))}
    env["CRAWLEE_PURGE_ON_START"] = "true" if purge else "false"
    code = WORKER_TPL.replace("@@BASE@@", repr(BASE)).replace("@@Q@@", repr(queue_name))
    proc = subprocess.run([sys.executable, "-X", "utf8", "-c", code], capture_output=True,
                          text=True, env=env, timeout=120)
    result = proc.stdout.strip() or proc.stderr.strip()[-110:]
    print("  CRAWLEE_PURGE_ON_START=%-6s -> %s" % (env["CRAWLEE_PURGE_ON_START"], result))


RESUME_WORKER = """
import asyncio, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from crawlee.crawlers import ParselCrawler
from crawlee.storages import RequestQueue

PORT = @@PORT@@
BASE = "http://127.0.0.1:%d" % PORT
HITS = []


class H(BaseHTTPRequestHandler):
    def do_GET(self):
        HITS.append(self.path)
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.end_headers()
        self.wfile.write(b"<html><body>ok</body></html>")

    def log_message(self, *a):
        pass


threading.Thread(target=lambda: ThreadingHTTPServer(("127.0.0.1", @@PORT@@), H).serve_forever(),
                 daemon=True).start()


async def run():
    crawler = ParselCrawler(configure_logging=False, max_requests_per_crawl=2,
                            request_manager=await RequestQueue.open(name="same-queue"))

    @crawler.router.default_handler
    async def handler(context):
        pass

    stats = (await crawler.run([BASE + "/a", BASE + "/b"])).to_dict()
    print("靶站被打 %d 次，统计里 finished=%d total=%d"
          % (len(HITS), stats["requests_finished"], stats["requests_total"]))


asyncio.run(run())
""".replace("@@BASE@@", repr(BASE)).replace("@@PORT@@", str(PORT + 3))


def run_resume(label: str, wipe: bool) -> None:
    if wipe:
        shutil.rmtree("storage_resume", ignore_errors=True)
    env = {k: v for k, v in os.environ.items() if not k.startswith(("CRAWLEE_", "APIFY_", "ACTOR_"))}
    env["CRAWLEE_PURGE_ON_START"] = "false"
    env["CRAWLEE_STORAGE_DIR"] = "./storage_resume"
    proc = subprocess.run([sys.executable, "-X", "utf8", "-c", RESUME_WORKER], capture_output=True,
                          text=True, env=env, timeout=120)
    print("  第%s次 -> %s" % (label, proc.stdout.strip() or proc.stderr.strip()[-110:]))


async def main() -> None:
    shutil.rmtree("storage", ignore_errors=True)
    await (await Dataset.open(name="keepme")).push_data({"tag": "preset"})
    print("\n3) 同一个 storage 目录连开三次进程，每次抓同一对 URL（队列名各换一个）：")
    run_worker(True, "pu-1")
    run_worker(True, "pu-2")
    run_worker(False, "pu-3")

    print("\n4) purge_on_start=false 时统计记录也会被沿用：同一份代码、同一个队列名连开两次")
    run_resume("一", wipe=True)
    run_resume("二", wipe=False)


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
1) Configuration 的全部字段（pydantic BaseSettings，字段名对应大写环境变量）：
  internal_timeout           默认=None                     环境变量=['crawlee_internal_timeout']
  default_browser_path       默认=None                     环境变量=['apify_default_browser_path', 'crawlee_default_browser_path']
  disable_browser_sandbox    默认=False                    环境变量=['apify_disable_browser_sandbox', 'crawlee_disable_browser_sandbox']
  log_level                  默认=INFO                     环境变量=['apify_log_level', 'crawlee_log_level']
  purge_on_start             默认=True                     环境变量=['apify_purge_on_start', 'crawlee_purge_on_start']
  persist_state_interval     默认=0:01:00                  环境变量=['apify_persist_state_interval_millis', 'crawlee_persist_state_interval_millis']
  system_info_interval       默认=0:00:01                  环境变量=['apify_system_info_interval_millis', 'crawlee_system_info_interval_millis']
  max_used_cpu_ratio         默认=0.95                     环境变量=['apify_max_used_cpu_ratio', 'crawlee_max_used_cpu_ratio']
  max_used_memory_ratio      默认=0.9                      环境变量=['apify_max_used_memory_ratio', 'crawlee_max_used_memory_ratio']
  max_event_loop_delay       默认=0:00:00.050000           环境变量=['apify_max_event_loop_delay_millis', 'crawlee_max_event_loop_delay_millis']
  max_client_errors          默认=1                        环境变量=['apify_max_client_errors', 'crawlee_max_client_errors']
  memory_mbytes              默认=None                     环境变量=['actor_memory_mbytes', 'apify_memory_mbytes', 'crawlee_memory_mbytes']
  available_memory_ratio     默认=0.25                     环境变量=['apify_available_memory_ratio', 'crawlee_available_memory_ratio']
  storage_dir                默认=./storage                环境变量=['apify_local_storage_dir', 'crawlee_storage_dir']
  headless                   默认=True                     环境变量=['apify_headless', 'crawlee_headless']
2) 环境变量单独起子进程实测（父进程环境保持干净）：
  CRAWLEE_STORAGE_DIR=./env_storage        读 storage_dir              -> {"storage_dir": "./env_storage"}
  CRAWLEE_LOG_LEVEL=DEBUG                  读 log_level                -> {"log_level": "DEBUG"}
  CRAWLEE_PURGE_ON_START=false             读 purge_on_start           -> {"purge_on_start": "False"}
  CRAWLEE_HEADLESS=false                   读 headless                 -> {"headless": "False"}
  CRAWLEE_MAX_CLIENT_ERRORS=7              读 max_client_errors        -> {"max_client_errors": "7"}
  CRAWLEE_INTERNAL_TIMEOUT=60              读 internal_timeout         ->     For further information visit https://errors.pydantic.dev/2.14/v/time_delta_parsing
  CRAWLEE_INTERNAL_TIMEOUT=PT60S           读 internal_timeout         -> {"internal_timeout": "0:01:00"}
  CRAWLEE_PERSIST_STATE_INTERVAL_MILLIS=5000 读 persist_state_interval   -> {"persist_state_interval": "0:00:05"}
  CRAWLEE_LOG_LEVEL=NOPE                   读 log_level                ->     For further information visit https://errors.pydantic.dev/2.14/v/literal_error
  CRAWLEE_DEDUPLICATION_POLICY=EXTENDED    读 log_level                -> {"log_level": "INFO"}
  CRAWLEE_MAX_CRAWLING_TIME=60             读 log_level                -> {"log_level": "INFO"}
  CRAWLEE_PROXY_URLS=http://127.0.0.1:8080 读 log_level                -> {"log_level": "INFO"}
  APIFY_LOCAL_STORAGE_DIR=./apify_storage  读 storage_dir              -> {"storage_dir": "./apify_storage"}
  ACTOR_MEMORY_MBYTES=2048                 读 memory_mbytes            -> {"memory_mbytes": "2048"}
3) 同一个 storage 目录连开三次进程，每次抓同一对 URL（队列名各换一个）：
  CRAWLEE_PURGE_ON_START=true   -> 默认数据集 2 条 ['/a', '/b']；具名 keepme 还剩 1 条
  CRAWLEE_PURGE_ON_START=true   -> 默认数据集 2 条 ['/a', '/b']；具名 keepme 还剩 1 条
  CRAWLEE_PURGE_ON_START=false  -> 默认数据集 4 条 ['/a', '/a', '/b', '/b']；具名 keepme 还剩 1 条
4) purge_on_start=false 时统计记录也会被沿用：同一份代码、同一个队列名连开两次
  第一次 -> 靶站被打 2 次，统计里 finished=2 total=2
  第二次 -> 靶站被打 0 次，统计里 finished=2 total=2
  • 环境变量名不是「字段名大写」这么简单：persist_state_interval 吃的是 CRAWLEE_PERSIST_STATE_INTERVAL_MILLIS，storage_dir 吃的是 CRAWLEE_STORAGE_DIR（APIFY 那个叫 APIFY_LOCAL_STORAGE_DIR）——表格里第二列是逐个实测出来的。
  • 时间类字段要 ISO8601 时长写法（PT60S），直接写 60 会 time_delta_parsing 报错；带 _MILLIS 后缀的变量收毫秒。
  • 非法值（CRAWLEE_LOG_LEVEL=NOPE）是构造时报错，不是静默忽略。
  • 而 CRAWLEE_DEDUPLICATION_POLICY、CRAWLEE_MAX_CRAWLING_TIME、CRAWLEE_PROXY_URLS 这些 JS 版才有的环境变量在 Python 版被完全忽略：Python 的 Configuration 只有这 15 个字段，多余的环境变量不报错也不生效。去重、限时、代理都得在代码里写。
  • purge_on_start（默认 True）只清默认存储，具名存储（例子里的 keepme）不会被清——这就是「用 RequestQueue.open(name=...) 做增量爬」能成立的根据。
  • 最后一段是个坑：purge_on_start=False 时，__CRAWLER_STATISTICS_0 这条统计记录也留着，于是第二次进程一睁眼就以为自己已经跑满了 max_requests_per_crawl，一个请求都不发（靶站被打 0 次）。要靠「不清存储 + 限流」做定时增量的话，记得别配 max_requests_per_crawl，或者每次换个队列名。

停一停再续爬

长跑任务中途要停，两种手段：crawler.stop()（体面收尾）和 max_requests_per_crawl（限量）。停了之后能不能接着爬，看队列里请求的状态：

stop_and_resume.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
# -*- coding: utf-8 -*-
"""中途停下（crawler.stop / max_requests_per_crawl / Ctrl-C），下次还能接着爬吗。"""

import asyncio
import shutil
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee._types import ConcurrencySettings
import os

os.environ["CRAWLEE_PURGE_ON_START"] = "false"
from crawlee.crawlers import ParselCrawler
from crawlee.storages import Dataset, RequestQueue

PORT = 8042
BASE = "http://127.0.0.1:%d" % PORT
INDEX = BASE + "/index"
HITS: list[str] = []


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        HITS.append(self.path)
        if self.path == "/index":
            links = "".join('<a href="/item/%d">i%d</a>' % (i, i) for i in range(1, 21))
            body = ("<html><body>%s<a href='/other'>x</a></body></html>" % links).encode()
        else:
            body = b"<html><body><p>detail</p></body></html>"
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()

STOP_AT = 3
handled: list[str] = []


async def new_crawler(queue_name: str, stop_after: int | None) -> ParselCrawler:
    crawler = ParselCrawler(
        configure_logging=False,
        concurrency_settings=ConcurrencySettings(min_concurrency=1, desired_concurrency=1,
                                                 max_concurrency=1),
        request_manager=await RequestQueue.open(name=queue_name))

    @crawler.router.default_handler
    async def handler(context) -> None:
        if context.request.url == INDEX:
            await context.enqueue_links()
        else:
            handled.append(context.request.url.replace(BASE, ""))
            await context.push_data({"item": context.request.url.replace(BASE, "/")})
        if stop_after is not None and len(handled) >= stop_after:
            crawler.stop("手动停：已经爬够 %d 条" % stop_after)

    return crawler


async def report(label: str) -> None:
    queue = await RequestQueue.open(name="resume")
    items = await (await Dataset.open()).list_items()
    print("  %-14s 队列 total=%d handled=%d pending=%d，数据集 %d 条"
          % (label, await queue.get_total_count(), await queue.get_handled_count(),
             (await queue.get_metadata()).pending_request_count, len(items)))


async def main() -> None:
    shutil.rmtree("storage", ignore_errors=True)
    first = await new_crawler("resume", STOP_AT)
    await first.run([INDEX])
    print("1) 爬到第 %d 条调用 crawler.stop()：已处理 %s" % (STOP_AT, sorted(handled)))
    await report("停止后")

    handled.clear()
    HITS.clear()
    second = await new_crawler("resume", None)
    await second.run([])
    print("2) 重开进程（purge_on_start=False）接着爬：又处理了 %d 条 %s"
          % (len(handled), sorted(handled)))
    await report("续爬后")
    print("   续爬这一轮靶站被打的次数:", len(HITS), "其中 /index 又请求了一次:",
          HITS.count("/index"))

    items = await (await Dataset.open()).list_items()
    tags = [i["item"] for i in items]
    print("3) 数据集里有没有重复:", [t for t in set(tags) if tags.count(t) > 1] or "无")

    # 4) 取出来没标 handled 的请求：在同进程里是 in-progress，得显式 reclaim 才会重排
    crash = await RequestQueue.open(name="crash")
    await crash.add_request(BASE + "/item/7")
    picked = await crash.fetch_next_request()
    meta = await crash.get_metadata()
    print()
    print("4) fetch_next_request 取出 /item/7 但不标 handled:",
          "total=%d pending=%d handled=%d"
          % (meta.total_request_count, meta.pending_request_count, meta.handled_request_count))
    print("   is_empty=%s is_finished=%s" % (await crash.is_empty(), await crash.is_finished()))
    handled.clear()
    HITS.clear()
    await crash.reclaim_request(picked)
    after = await new_crawler("crash", None)
    await after.run([])
    print("   reclaim_request 之后再 run：重新抓到 %s，靶站 /item/7 命中 %d 次"
          % (sorted(handled), HITS.count("/item/7")))


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
1) 爬到第 3 条调用 crawler.stop()：已处理 ['/item/1', '/item/2', '/item/3']
  停止后            队列 total=22 handled=4 pending=18，数据集 3 条
2) 重开进程（purge_on_start=False）接着爬：又处理了 18 条 ['/item/10', '/item/11', '/item/12', '/item/13', '/item/14', '/item/15', '/item/16', '/item/17', '/item/18', '/item/19', '/item/20', '/item/4', '/item/5', '/item/6', '/item/7', '/item/8', '/item/9', '/other']
  续爬后            队列 total=22 handled=22 pending=0，数据集 21 条
   续爬这一轮靶站被打的次数: 18 其中 /index 又请求了一次: 0
3) 数据集里有没有重复: 无
4) fetch_next_request 取出 /item/7 但不标 handled: total=1 pending=1 handled=0
   is_empty=True is_finished=False
   reclaim_request 之后再 run：重新抓到 ['/item/7']，靶站 /item/7 命中 1 次
  • 第 1 段爬到第 3 条调用 crawler.stop()：total=22（列表页 + 20 条详情 + 一个外域链接）、handled=4、pending=18，没抓的 18 条原样躺在队列里。
  • 第 2 段换新爬虫、同一个队列名、CRAWLEE_PURGE_ON_START=false 再 run([])：不重新下发种子，直接消费队列里剩下的 18 条，最后 handled=22。
  • 「续爬」不会重放列表页（/index 命中数 0），因为它是 handled 状态；也不会重复入库（数据集 21 条无重复）。
  • 第 4 段是「取出来没标 handled」的下场：is_empty=True 但 is_finished=False，同进程里再 run([]) 不会碰它，必须显式 reclaim_request 才重新排队。真实崩溃时新进程的队列客户端会把上次 in-progress 的请求回收重试，所以同一条内容可能被抓两次——入库要靠 uniqueKey 或业务主键兜底。

内存存储与脚手架

本地跑测试不想产生 storage/ 目录，可以整体换成内存存储；起新项目则可以用官方 CLI 生成骨架：

memory_and_cli.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
# -*- coding: utf-8 -*-
"""把存储换成纯内存，以及官方 CLI 的 crawlee create 生成的项目骨架。"""

import asyncio
import subprocess
import sys
from pathlib import Path

from crawlee.crawlers import ParselCrawler
from crawlee.storage_clients import MemoryStorageClient
from crawlee.storages import Dataset, KeyValueStore, RequestQueue

print("1) MemoryStorageClient：跑完磁盘上什么都没有")
client = MemoryStorageClient()


async def memory() -> None:
    crawler = ParselCrawler(configure_logging=False, storage_client=client,
                            max_requests_per_crawl=1)

    @crawler.router.default_handler
    async def handler(context) -> None:
        await context.push_data({"title": context.selector.css("title::text").get() or "无标题"})

    await crawler.run(["https://ssr1.scrape.center/page/1"])
    items = await (await crawler.get_dataset()).list_items()
    print("   抓到的条数:", len(items), items[:1])
    print("   磁盘上有没有 storage 目录:", Path("storage").exists())
    same = await Dataset.open(storage_client=client)
    print("   同一个 client 再 open 一次:", len(await same.list_items()), "条（数据还在内存里）")
    other = await Dataset.open(storage_client=MemoryStorageClient())
    print("   换一个 MemoryStorageClient（等价于重开进程）:", len(await other.list_items()), "条")

    queue = await RequestQueue.open(storage_client=client)
    await queue.add_request("https://example.com/a")
    await queue.add_request("https://example.com/a")
    print("   内存队列 add 两次同一 URL -> total=%d" % await queue.get_total_count())
    kvs = await KeyValueStore.open(storage_client=client)
    await kvs.set_value("cfg", {"n": 1})
    print("   内存 KVS 读回:", await kvs.get_value("cfg"))


asyncio.run(memory())

print("\n2) 官方 CLI（pip install 'crawlee[cli]' 装出来的 crawlee 命令）")


def run_cli(args: list[str]) -> None:
    exe = str(Path(sys.executable).parent / "crawlee.exe")
    proc = subprocess.run([exe] + args, capture_output=True,
                          text=True, timeout=180, encoding="utf-8", errors="replace")
    out = ((proc.stdout or "") + (proc.stderr or "")).strip().splitlines()
    print("   $ crawlee %s -> 退出码 %d，输出 %d 行" % (" ".join(args), proc.returncode, len(out)))
    for line in out[:8]:
        print("     | " + line[:110])


run_cli(["--version"])
run_cli(["--help"])
run_cli(["create", "demo-parsel", "--crawler-type", "parsel", "--http-client", "httpx",
         "--package-manager", "uv", "--start-url", "https://ssr1.scrape.center",
         "--no-apify", "--no-install"])

root = Path("demo-parsel")
if root.exists():
    print("   生成的文件:", sorted(p.relative_to(root).as_posix() for p in root.rglob("*")
                                if p.is_file()))
    main_py = (root / "demo_parsel" / "main.py").read_text(encoding="utf-8")
    print("   main.py 前 6 行:")
    for line in main_py.splitlines()[:6]:
        print("     | " + line)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
1) MemoryStorageClient：跑完磁盘上什么都没有
   抓到的条数: 1 [{'title': 'Scrape | Movie'}]
   磁盘上有没有 storage 目录: False
   同一个 client 再 open 一次: 1 条（数据还在内存里）
   换一个 MemoryStorageClient（等价于重开进程）: 1 条
   内存队列 add 两次同一 URL -> total=2
   内存 KVS 读回: {'n': 1}
2) 官方 CLI（pip install 'crawlee[cli]' 装出来的 crawlee 命令）
   $ crawlee --version -> 退出码 0，输出 1 行
     | 1.11.1
   $ crawlee --help -> 退出码 0，输出 14 行
     | Usage: crawlee [OPTIONS] COMMAND [ARGS]...                                    
     |                                                                                
     |  Crawlee is a web scraping and browser automation library.                     
     |                                                                                
     | +- Options -------------------------------------------------------------------+
     | | --version             -V        Print Crawlee version                       |
     | | --install-completion            Install completion for the current shell.   |
     | | --show-completion               Show completion for the current shell, to   |
   $ crawlee create demo-parsel --crawler-type parsel --http-client httpx --package-manager uv --start-url https://ssr1.scrape.center --no-apify --no-install -> 退出码 0，输出 3 行
     | Your project "demo-parsel" was created.
     | To run it, navigate to the directory: "cd demo-parsel", install the project using "uv sync", and run it using 
     | See the "demo-parsel/README.md" for more information.
   生成的文件: ['.dockerignore', 'Dockerfile', 'README.md', 'demo_parsel/__init__.py', 'demo_parsel/__main__.py', 'demo_parsel/__pycache__/__init__.cpython-314.pyc', 'demo_parsel/__pycache__/main.cpython-314.pyc', 'demo_parsel/__pycache__/routes.cpython-314.pyc', 'demo_parsel/main.py', 'demo_parsel/routes.py', 'pyproject.toml']
   main.py 前 6 行:
     | from crawlee.crawlers import ParselCrawler
     | from crawlee.http_clients import HttpxHttpClient
     | from .routes import router
     | 
     | async def main() -> None:
     |     """The crawler entry point."""
  • storage_client=MemoryStorageClient() 之后磁盘上一个文件都没有，数据只活在这个 client 对象里：同一个 client 再 open 一次还在，换一个 client 就等价于重开进程（本机上第二个 client 里读到的还是 1 条，因为 Dataset.open() 不带 client 时读的是全局默认 client——真要每次干净，就连 client 一起新建）。
  • crawlee 这个命令由 pip install 'crawlee[cli]' 带出来，只有一个子命令 create；crawlee run / check / purge 这些（JS 版的）命令在 Python 版不存在。
  • create 的五个选项能全程非交互：--crawler-type（beautifulsoup/parsel/adaptive-/playwright-）、--http-client（impit/httpx/curl-impersonate）、--package-manager（poetry/pip/uv）、--start-url、--apify/--no-apify，不给就会进交互式选择。
  • 生成的是 uv 工程（pyproject.toml 里依赖写成 crawlee[parsel,httpx]）、代码切成 main.py + routes.py，还带一个 Dockerfile（基镜像 apify/actor-python:3.13）。跑起来是 uv sync 然后 uv run python -m demo_parsel。

三个框架怎么选

Scrapy、feapder、crawlee 三个都在本机装过、跑过，写法上的差别是这样：

要做的事 Scrapy feapder Crawlee
装 pip install scrapy pip install feapder pip install 'crawlee[parsel]'
起步 建工程 + items + pipelines + settings 四处 建 Spider 子类 + Item 一个 ParselCrawler + 一个 handler
解析 selector（CSS/XPath） 自己选（正则/selector） 三种：bytes / parsel selector / soup
跟链接 CrawlSpider 的 rule 或 response.follow_all start_requests 里自己拼 context.enqueue_links(strategy=...)
去重 默认 RFPDupeFilter，只记指纹 任务表 / Redis 游标 队列主键 uniqueKey（归一化 URL）
断点续爬 靠 JOBDIR + dupefilter 落盘 MySQL/Redis 任务表 storage/ 目录里的队列 + purge_on_start=False
并发 CONCURRENT_REQUESTS 定值 SPIDER_THREAD_COUNT 定值 ConcurrencySettings 三个数 + 自动伸缩
浏览器 要接 scrapy-playwright 有 BrowserSpider（Selenium/Playwright 两套） 原生 PlaywrightCrawler，还有自适应版
代理 ProxyMiddleware + ROTATING_PROXY_LIST 自带代理池（Redis） ProxyConfiguration，支持分档自动升档
入库 Item Pipeline（写几行代码） Item Pipeline + 批量化接口 默认只进 storage/datasets，外部库自己写

选谁的建议：

  • 要「入库、去重、定时、报警」这一整套调度，feapder 更省事，它把任务表和批量化做进了框架（见《爬虫-Feapder》）。
  • 要高度定制的管道、中间件生态、大量现成扩展，Scrapy 仍是首选（见《爬虫-Scrapy》），代价是工程结构固定，改起来要动 settings。
  • 要「一个脚本就能跑」＋ 静态/浏览器自动切换 ＋ 反封禁（代理分档、会话轮换、TLS 指纹伪装），crawlee 最顺手，它是 asyncio 一路写到底，API 都在对象上而不是配置项里。
  • crawlee 是 requests/httpx 这一路的「框架化」版本，学习曲线比 Scrapy 平；但它默认不写数据库，落地环节要自己接（context.push_data 只写到本地数据集）。
  • 依赖体积也是实际差别：crawlee 裸装 21 个包就够用，装 [all] 会拉到 127 个包，Playwright 那条路还要再 playwright install chromium。

常见报错速查

这一节全是本机复现过的报错，原文照抄。看懂这几句基本就能定位问题：

error_cheatsheet.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
# -*- coding: utf-8 -*-
"""常见报错速查：每一条都在本机跑过，报错原文照抄（靶站在 8043 端口）。"""

import asyncio
import logging
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from crawlee import Request
from crawlee._types import ConcurrencySettings
from crawlee.crawlers import HttpCrawler, ParselCrawler, PlaywrightCrawler
from crawlee.storages import RequestQueue

logging.getLogger().setLevel(logging.CRITICAL)

PORT = 8043
BASE = "http://127.0.0.1:%d" % PORT


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = b'<html><body><a href="/p/1">1</a><a href="/p/2">2</a></body></html>'
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()


async def crawl_once(crawler, urls) -> str:
    """跑一轮，把失败原因原样抓回来（run() 本身不抛异常，异常记在统计里）。"""
    caught = []

    @crawler.failed_request_handler
    async def on_failed(context, error) -> None:
        first = str(error).splitlines()[0].split(" at ")[0]
        caught.append("%s: %s" % (type(error).__name__, first[:130]))

    await crawler.run(urls)
    return caught[0] if caught else "跑完了，没有失败"


async def main() -> None:
    print("1) 直接 Request(url=...) 构造：")
    try:
        Request(url=BASE)
    except Exception as exc:
        print("   %s: %s" % (type(exc).__name__, " | ".join(str(exc).splitlines()[1:3])))
    print("   改成 Request.from_url(...) 就好:", Request.from_url(BASE).unique_key)

    print("\n2) 只用 router.handler('detail') 而种子不带标签：")
    label_crawler = ParselCrawler(configure_logging=False, max_requests_per_crawl=1,
                                  request_manager=await RequestQueue.open(name="e-label"))

    @label_crawler.router.handler("detail")
    async def on_detail(context) -> None:
        pass

    print("  ", await crawl_once(label_crawler, [BASE]))

    print("\n3) enqueue_links 的策略名写错（大写 A）：")
    bad_strategy = ParselCrawler(configure_logging=False, max_requests_per_crawl=1,
                                 request_manager=await RequestQueue.open(name="e-strategy"))

    @bad_strategy.router.default_handler
    async def on_page(context) -> None:
        await context.enqueue_links(strategy="All")

    print("  ", await crawl_once(bad_strategy, [BASE]))

    print("\n4) HttpCrawler 的 handler 里想用 context.selector：")
    plain = HttpCrawler(configure_logging=False, max_requests_per_crawl=1,
                        request_manager=await RequestQueue.open(name="e-attrs"))

    @plain.router.default_handler
    async def on_plain(context) -> None:
        try:
            context.selector  # noqa: B018
        except AttributeError as exc:
            print("   AttributeError: %s" % exc)

    await plain.run([BASE])

    print("\n5) 存储名带下划线：")
    try:
        await RequestQueue.open(name="my_queue")
    except Exception as exc:
        print("   %s: %s" % (type(exc).__name__, exc))

    print("\n6) ConcurrencySettings 只降 max 不降 desired：")
    try:
        HttpCrawler(configure_logging=False,
                    concurrency_settings=ConcurrencySettings(max_concurrency=1))
    except Exception as exc:
        print("   %s: %s" % (type(exc).__name__, exc))

    print("\n7) browser_type 指的浏览器没装（本机只装了 chromium）：")
    webkit = PlaywrightCrawler(configure_logging=False, max_requests_per_crawl=1,
                               browser_type="webkit",
                               request_manager=await RequestQueue.open(name="e-browser"))

    @webkit.router.default_handler
    async def on_webkit(context) -> None:
        pass

    print("  ", await crawl_once(webkit, [BASE]))

    print("\n8) 把 run() 的返回值当字典下标：")
    stats_crawler = HttpCrawler(configure_logging=False, max_requests_per_crawl=1,
                                request_manager=await RequestQueue.open(name="e-stats"))

    @stats_crawler.router.default_handler
    async def on_stats(context) -> None:
        pass

    stats = await stats_crawler.run([BASE])
    try:
        stats["requests_finished"]
    except Exception as exc:
        print("   %s: %s" % (type(exc).__name__, exc))
        print("   正确写法 stats.to_dict()['requests_finished'] =", stats.to_dict()["requests_finished"])

    print("\n9) parsel 选择器取不到东西（不抛异常，给空集合）：")
    miss = ParselCrawler(configure_logging=False, max_requests_per_crawl=1,
                         request_manager=await RequestQueue.open(name="e-miss"))

    @miss.router.default_handler
    async def on_miss(context) -> None:
        print("   css('.nothing') -> %r，.get() -> %r，.attrib -> %r"
              % (context.selector.css(".nothing"), context.selector.css(".nothing").get(),
                 context.selector.css(".nothing").attrib))

    await miss.run([BASE])


asyncio.run(main())
server.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
1) 直接 Request(url=...) 构造：
   ValidationError: uniqueKey |   Field required [type=missing, input_value={'url': 'http://127.0.0.1:8043'}, input_type=dict]
   改成 Request.from_url(...) 就好: http://127.0.0.1:8043
2) 只用 router.handler('detail') 而种子不带标签：
   RuntimeError: No handler matches label `None` and no default handler is configured
3) enqueue_links 的策略名写错（大写 A）：
   AssertionError: Expected code to be unreachable, but got: 'All'
4) HttpCrawler 的 handler 里想用 context.selector：
   AttributeError: 'ParsedHttpCrawlingContext' object has no attribute 'selector'
5) 存储名带下划线：
   ValueError: Invalid storage name "my_queue". Name can only contain letters "a" through "z", the digits "0" through"9", and the hyphen ("-") but only in the middle of the string (e.g. "my-value-1")
6) ConcurrencySettings 只降 max 不降 desired：
   ValueError: desired_concurrency cannot be greater than max_concurrency
7) browser_type 指的浏览器没装（本机只装了 chromium）：
   Error: BrowserType.launch_persistent_context: Executable doesn't exist
8) 把 run() 的返回值当字典下标：
   TypeError: 'FinalStatistics' object is not subscriptable
   正确写法 stats.to_dict()['requests_finished'] = 1
9) parsel 选择器取不到东西（不抛异常，给空集合）：
   css('.nothing') -> []，.get() -> None，.attrib -> {}

排查顺序建议：

  • 报 unexpected keyword argument，先怀疑自己抄的是 JS 版或旧版文档；用 inspect.signature(类.__init__) 打一遍参数最快（本文各处都是这么打的）。
  • 报 No handler matches label，是标签路由：要么种子带上 label，要么配 @crawler.router.default_handler 兜底。
  • 什么都不抓、统计里 requests_total=0，八成是默认队列记得这些 URL：换个具名队列，或删掉 storage/。
  • requests/httpx 习惯里的 headers=、verify= 在爬虫构造函数上都不存在，前者写进 Request.from_url(..., headers=...)，后者属于 HTTP 客户端。

参考