Scrapling Spider 快速上手指南：从第一个爬虫到数据导出与抓取控制

本文基于 Scrapling 仓库中的 Spider 入门文档编写，带你从零搭建一个可运行的爬虫：定义 name、start_urls 与 parse() 三大要素，用 start() 一键运行并获取详细的 CrawlResult 统计，再掌握 response.follow() 跨页跟随、ItemList 多格式数据导出、allowed_domains 域过滤与 robots_txt_obey 合规抓取。读完你应能独立编写、运行并控制一个多页爬虫，同时理解 Scrapling 引擎在每个环节背后的真实实现。

你的第一个 Spider

Spider 是一个类，用于定义如何从网站抓取与提取数据。最简单的 Spider 如下：

from scrapling.spiders import Spider, Response

class QuotesSpider(Spider):
    name = "quotes"
    start_urls = ["https://quotes.toscrape.com"]

    async def parse(self, response: Response):
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(""),
                "author": quote.css("small.author::text").get(""),
            }

每个 Spider 需要三样东西：

1. name：Spider 的唯一标识符。
2. start_urls：开始爬取的 URL 列表。
3. parse()：一个异步生成器方法，处理每个响应并 yield 结果。

parse() 方法处理每个响应。你使用的是与 Scrapling 的 Selector/Response 对象 完全相同的选取方法，然后通过 yield 字典来输出抓取到的条目。

源码视角：Spider 基类到底提供了什么

从 Spider 基类 的定义可以看出，parse() 是抽象方法，且类型签名明确要求它是异步生成器（AsyncGenerator），这与"所有回调都必须是 async def + yield"的要求一致。基类还内置了一批类级默认配置，写 Spider 时可以直接覆盖：

配置项	默认值	作用
concurrent_requests	4	全局并发请求数
concurrent_requests_per_domain	0	单域并发限制，0 表示仅用全局限制
download_delay	0.0	请求间延迟（秒）
robots_txt_obey	False	是否遵守 robots.txt
autothrottle_enabled	False	是否启用自动限速
max_blocked_retries	3	被拦截请求的最大重试次数
logging_level	logging.DEBUG	日志级别

两个初始化细节值得注意：

• 如果 name 为 None，__init__ 会直接抛出 ValueError（spider.py L112-L113），所以忘记命名会立刻暴露问题。
• 每个 Spider 实例都会配置一个名为 scrapling.spiders.<name> 的独立 logger，日志格式中带上 Spider 名，这就是"运行期间一切都会打印到终端"的来源（spider.py L115-L136）。

在引擎侧，CrawlerEngine._run_callbacks 会消费回调 yield 出来的对象：遇到 Request 就（经过域过滤后）入队，遇到 dict 就交给 on_scraped_item 钩子处理后计入条目——也就是说，parse() 里混着 yield 字典和新请求是被明确支持的两种结果类型。

运行 Spider：start() 与 CrawlResult

要运行 Spider，创建实例并调用 start()：

result = QuotesSpider().start()

start() 方法在内部处理了所有异步机制，因此你无需操心事件循环。Spider 运行期间，发生的一切都会记录到终端，爬取结束时你会得到非常详细的统计信息。

这些统计位于返回的 CrawlResult 对象中，它提供了你需要的一切：

result = QuotesSpider().start()

# 访问抓取的条目
for item in result.items:
    print(item["text"], "-", item["author"])

# 查看统计信息
print(f"Scraped {result.stats.items_scraped} items")
print(f"Made {result.stats.requests_count} requests")
print(f"Took {result.stats.elapsed_seconds:.1f} seconds")

# 爬虫是正常结束还是被暂停了？
print(f"Completed: {result.completed}")

源码视角：start() 的内部机制

从 start() 的实现 可以看到：

• 它通过 anyio.run(self.__run, backend="asyncio", ...) 同步封装了整套异步爬取流程，对外表现为一次阻塞调用；还支持 use_uvloop=True 参数以启用更快的 uvloop/winloop 事件循环（若可用）。
• 它临时安装了 SIGINT 处理器：按一次 Ctrl+C 发起优雅暂停（等待活动任务完成），再按一次强制停止；若配置了 crawldir，优雅关闭时还会保存 checkpoint 以便后续恢复。
• __run 中创建 CrawlerEngine、执行 engine.crawl() 得到统计，再把 engine.items 与 paused 标志打包成 CrawlResult 返回。

CrawlResult 本身很轻：stats（CrawlStats 实例）、items（ItemList）、paused 布尔值，其中 completed 属性即 not paused。它同时实现了 __len__ 与 __iter__，可以直接 len(result) 或对结果迭代条目。

而终端打印的那些统计，全部来自 CrawlStats 数据类，除文档示例用到的 items_scraped、requests_count、elapsed_seconds 外，还可直接访问：failed_requests_count（失败请求）、offsite_requests_count（被域过滤丢弃的请求）、robots_disallowed_count（被 robots.txt 拦截的请求）、cache_hits/cache_misses（缓存命中情况）、response_bytes（响应总字节）、requests_per_second（吞吐量，派生属性）、response_status_count（按 HTTP 状态码的计数）等。stats.to_dict() 会把这些汇总成可序列化字典，elapsed_seconds 与 requests_per_second 均由 start_time/end_time 派生（result.py L149-L157）。

跟随链接：response.follow()

大多数爬取需要跨多页跟随链接。使用 response.follow() 创建后续请求：

from scrapling.spiders import Spider, Response

class QuotesSpider(Spider):
    name = "quotes"
    start_urls = ["https://quotes.toscrape.com"]

    async def parse(self, response: Response):
        # 从当前页提取条目
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(""),
                "author": quote.css("small.author::text").get(""),
            }

        # 跟随 "下一页" 链接
        next_page = response.css("li.next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

response.follow() 会自动处理相对 URL，将其与当前页面 URL 拼接；同时默认把当前页面设置为 Referer 请求头。

你也可以把后续请求指向不同的回调方法，以区分不同类型的页面：

async def parse(self, response: Response):
    for link in response.css("a.product-link::attr(href)").getall():
        yield response.follow(link, callback=self.parse_product)

async def parse_product(self, response: Response):
    yield {
        "name": response.css("h1::text").get(""),
        "price": response.css(".price::text").get(""),
    }

注意： 所有回调方法都必须是异步生成器（使用 async def 和 yield）。

源码视角：follow() 的完整参数

Response.follow 的实现 签名比文档示例展示的更丰富，除 url 和 callback 外还提供：

参数	说明
sid	指定用哪个会话发起该请求，留空则沿用上一请求的会话
priority	优先级数值，越大越先被处理
dont_filter	该请求若此前执行过，禁用去重过滤器允许再次执行
meta	附加到请求上的元数据字典
referer_flow	默认 True，将当前响应 URL 作为新请求的 referer
**kwargs	透传给会话的额外请求参数（如 headers、data）

两个实现细节：

• 参数继承：follow() 会把上一请求的会话参数与新传入的参数合并（新值优先）（custom.py L120-L121），所以连续 yield response.follow(...) 时 headers、代理等设置会自动流转。
• 前置条件：follow() 要求 response.request 已由引擎设置，否则抛出 TypeError（custom.py L117-L118）——这意味着 follow() 只适用于引擎回调上下文中的响应，不能对裸构造的 Response 使用。

生成的 Request 对象 内部通过 update_fingerprint() 计算 SHA1 指纹（涵盖会话 ID、方法、规范化 URL 与请求体）用于去重，相同请求不会重复入队；priority 则通过 __lt__/__gt__ 的比较实现参与调度排序（request.py L71-L143）。

导出数据：ItemList 的内置导出方法

result.items 返回的 ItemList 带有内置导出方法：

result = QuotesSpider().start()

# 导出为 JSON
result.items.to_json("quotes.json")

# 导出为带缩进的 JSON
result.items.to_json("quotes.json", indent=True)

# 导出为 JSON Lines（每行一个 JSON 对象）
result.items.to_jsonl("quotes.jsonl")

# 导出为 CSV 或 XML
result.items.to_csv("quotes.csv")
result.items.to_xml("quotes.xml")

以上方法都会在父目录不存在时自动创建（源码中每个方法都先执行 Path(path).parent.mkdir(parents=True, exist_ok=True)）。

to_csv() 会为所有条目中出现过的键各写一列，因此即使条目键不完全一致也能导出，缺失单元格留空。可以传 fields=[...] 自行挑选列及其顺序，或 delimiter="\t" 生成 TSV 文件。to_xml() 把每个条目包在 <item> 元素内，整体再包在 <items> 根元素内，两者可通过 root_tag 和 item_tag 改名。

两种格式下，非简单标量值（嵌套字典或列表）都会被写成 JSON，确保没有任何数据被静默丢弃。

源码视角：导出的容错处理

从 ItemList 的实现 可以确认上述行为并补充几点：

• 序列化统一使用 orjson（带 OPT_SERIALIZE_NUMPY 选项），to_json(indent=True) 使用 2 空格缩进（源码注释标明会稍慢一些）。
• CSV 列的默认顺序是"键在条目中出现的先后顺序"（用字典推导保序去重，result.py L77），to_csv 内部使用 csv.DictWriter 且 extrasaction="ignore"。
• XML 导出比文档描述更健壮：条目键若不是合法的 XML 标签名，会被改写为合法形式并把原始键名保留在 name 属性中；文本中的非法 XML 控制字符会被剔除（result.py L17-L19, L89-L118）。非标量值序列化为 JSON 的逻辑集中在 _stringify()（result.py L22-L28），CSV 与 XML 共用。
• 每次导出都会以 INFO 级别记录"Saved N items to

过滤域名：allowed_domains

使用 allowed_domains 将 Spider 限制在特定域名内，防止它意外跟随到外部网站的链接：

class MySpider(Spider):
    name = "my_spider"
    start_urls = ["https://example.com"]
    allowed_domains = {"example.com"}

    async def parse(self, response: Response):
        for link in response.css("a::attr(href)").getall():
            # 指向其他域名的链接会被静默丢弃
            yield response.follow(link, callback=self.parse)

子域名会被自动匹配，因此设置 allowed_domains = {"example.com"} 同样允许 sub.example.com、blog.example.com 等。

被过滤掉的请求会计入 stats.offsite_requests_count，方便你看到有多少请求被丢弃。

源码视角：域匹配与计数发生在哪

这两个行为都能在 CrawlerEngine 中逐行对应：

• _is_domain_allowed() 的判断逻辑是 domain == allowed or domain.endswith("." + allowed)——这就是"子域名自动匹配"的实现；若 allowed_domains 为空集则全部放行。
• 在 _run_callbacks 中，回调 yield 出的 Request 若未通过域检查，则 self.stats.offsite_requests_count += 1 并打印一条 DEBUG 日志（Filtered offsite request to: ...），请求不进入调度器——即"静默丢弃且可计数"。

robots.txt 合规：robots_txt_obey

设置 robots_txt_obey = True，让 Spider 在抓取任何域名前先遵守 robots.txt 规则：

class PoliteSpider(Spider):
    name = "polite"
    start_urls = ["https://example.com"]
    robots_txt_obey = True

    async def parse(self, response: Response):
        for link in response.css("a::attr(href)").getall():
            yield response.follow(link, callback=self.parse)

启用后，Spider 会：

1. 预取 robots.txt：爬取开始前，并发抓取 start_urls 中所有域名的 robots.txt。
2. 检查每个请求：对照该域名 robots.txt 的 Disallow 规则检查。被禁止的请求被静默丢弃，并计入 stats.robots_disallowed_count。
3. 遵守 Crawl-delay 与 Request-rate 指令：取指令值与你配置的 download_delay 的较大者。这意味着 robots.txt 的延迟永远不会降低你配置的延迟，只会在需要时增大它。

robots.txt 使用 Spider 的默认会话抓取，并在整个爬取期间按域缓存。爬取中途发现的域名（不在 start_urls 中）会在首次请求该域名时抓取其 robots.txt。

注意： robots_txt_obey 默认关闭。它不影响你的并发设置——只调整请求之间的延迟。

如果不想手工设定 download_delay，可以设置 autothrottle_enabled = True，Spider 会根据各域名的响应速度自行调节延迟，被拦截时自动退避。参见 AutoThrottle 进阶文档。

源码视角：RobotsTxtManager 与延迟解析

上述三步流程对应两处源码：

• RobotsTxtManager 负责 robots.txt 的抓取、解析与缓存：prefetch() 用 anyio 的 create_task_group 并发预热 start_urls 涉及的各域；can_fetch() 以通配 user-agent * 判定 URL 是否允许抓取（基于 protego 库解析）；解析器按 domain 缓存在 self._cache 中，整个爬取期间只取一次。抓取或解析失败只记录 WARNING 并回退到空规则，不会中断爬取。
• 延迟取最大值的逻辑在 CrawlerEngine._get_domain_delay：它同时读取 Crawl-delay 与 Request-rate 两类指令，其中 Request-rate 会被换算为 周期 / 请求数 再与 download_delay 取最大值，结果按域名缓存在 _domain_delays 中。源码注释明确说明：对于预取过的域名这是本地解析器读取，而爬取中途发现的域名会在此处按需抓取——与文档描述一致。
• robots_txt_obey 的默认值 False 定义在 Spider 类属性 中；只有开启它，引擎才会初始化 robots 管理器与按域延迟缓存（engine.py L80-L81）。

小结与延伸阅读

到这里，你已经覆盖了入门文档的全部主线：定义三要素 Spider → start() 运行并读取 CrawlResult → response.follow() 跨页跟随 → ItemList 四种格式导出 → allowed_domains 域过滤 → robots_txt_obey 合规抓取。所有行为均与 scrapling/spiders 目录下的源码实现一一对应，相关测试用例（如 tests/spiders/test_spider.py、tests/spiders/test_robotstxt.py、tests/spiders/test_result.py）可用于进一步验证。

继续深入可阅读仓库文档：

• Spider 进阶：AutoThrottle、开发模式等
• Spider 架构详解
• 会话管理
• 选择器与主要解析类
• 抓取器选择指南与 Response 对象
• Scrapy 集成说明