爬虫-Scrapy

发表于 2019-12-24 更新于 2026-11-07 分类于 Python Python教程爬虫本文字数： 69k 阅读时长 ≈ 2:06

这篇讲什么

全部示例在本机重跑过：Python 3.14.10 + Scrapy 2.19.0 + Twisted 26.10.0 + parsel 1.12.1 + itemloaders 1.5.0。抓取主线换成公开练习站（服务端渲染，列表页/详情页都有数据），接口线用；中间件、代理、图片防盗链、落库仍打本机靶站 http://127.0.0.1:8010 和本机数据库（那是可控条件，在线站给不了）。输出逐字贴出。
主线是「工程骨架 → Request/Response/Selector/Item/ItemLoader → 回调链 → Pipeline → 中间件 → 图片文件 → 代理登录 → 存储 → 分布式」，每一环都有可复制的代码和真实日志。
Scrapy 2.x 相对旧教程的变化不少：scrapy.contrib 没了、Spider.start_requests() 没了、start() 改成异步生成器（普通 def start 直接报错）、shell 里的 sel 没了、SelectorList.first 没了，本篇把这些坑连报错原文一起列出来。
篇幅优先给「能跑的写法」，配置项只讲实测影响过结果的那些，细节以官方文档为准。

装与版本

1	pip install scrapy

装完先确认版本，注意 -a 和 -v 不一样：本机 scrapy version -a 只打印一行，要依赖清单得用 -v。

1 2	scrapy version -a scrapy version -v

D:\...\Scripts>scrapy.exe version -a
Scrapy 2.19.0

D:\...\Scripts>scrapy.exe version -v
Scrapy       : 2.19.0
lxml         : 6.1.3
libxml2      : 2.11.9
cssselect    : 1.6.0
parsel       : 1.12.1
w3lib        : 2.6.0
Twisted      : 26.10.0
Python       : 3.14.10 | packaged by Anaconda, Inc. | (main, Oct 27 2026, 01:09:10) [MSC v.1942 64 bit (AMD64)]
pyOpenSSL    : 26.4.0 (OpenSSL 4.0.3 29 Sep 2026)
cryptography : 50.0.2
Platform     : Windows-11-10.0.26300-SP0

同一份信息每次启动都会打一遍 DEBUG 日志（scrapy.utils.log 的 Versions: 那行），排查环境问题时看日志比敲命令快。
scrapy 这个命令本身也可以用 python -m scrapy 代替，本机两种都能跑，输出一致。

1	python -m scrapy version -a

1 2	D:\...\Scripts>python -m scrapy version -a Scrapy 2.19.0

老写法的报错原文

网上大量 2015 年前后的教程还在写 scrapy.contrib.*，现在一律是 ModuleNotFoundError，本机逐条实测：

1	python # 进 REPL，逐条敲下面这几行 2015 年教程里的 import

>>> import scrapy.contrib
ModuleNotFoundError: No module named 'scrapy.contrib'

>>> from scrapy.contrib.linkextractors import LinkExtractor
ModuleNotFoundError: No module named 'scrapy.contrib'

>>> from scrapy.contrib.spiders import CrawlSpider, Rule
ModuleNotFoundError: No module named 'scrapy.contrib'

>>> from scrapy.contrib.loader import ItemLoader
ModuleNotFoundError: No module named 'scrapy.contrib'

现在的正确位置：LinkExtractor 在 scrapy.linkextractors，CrawlSpider / Rule 在 scrapy.spiders，ItemLoader 在 scrapy.loader。
另一个容易踩的：处理器模块 scrapy.loader.processors 在本机 2.19.0 上也不存在了，处理器要从 itemloaders 拿。

1	python # 同样在 REPL 里逐条试下面 scrapy 顶层 / 子模块的写法

>>> from scrapy import ItemLoader
ImportError: cannot import name 'ItemLoader' from 'scrapy' (D:\...\scrapy\__init__.py)

>>> import scrapy.loader.processors
ModuleNotFoundError: No module named 'scrapy.loader.processors'

>>> from scrapy.loader import ItemLoader
>>> ItemLoader
<class 'scrapy.loader.ItemLoader'>

同理，parsel 顶层也没有 SelectorSyntaxError，选择器写错时抛的是 SelectorError／ValueError 之类，别按名字硬 import。

1	python # REPL 里确认 parsel 顶层到底导出了哪些名字

1 2	>>> from parsel import SelectorSyntaxError ImportError: cannot import name 'SelectorSyntaxError' from 'parsel' (D:\...\parsel\__init__.py)

工程骨架

用 scrapy startproject 生成骨架，命令会连着建出「外层工程目录 + 同名 Python 包」两层。

Scrapy 架构

1	scrapy startproject scrapyproj

New Scrapy project 'probe_new_proj', using template directory 'D:\...\scrapy\templates\project', created in:
     E:\...\_tmp_dir\probe_new_proj

You can start your first spider with:
     cd probe_new_proj
     scrapy genspider example example.com

目录结构（本机跑完所有示例之后的样子，注释写在右边）：

1 2	scrapy startproject scrapyproj # 生成下面这棵树（spiders/ 里的 books.py 是后面手写的） find scrapyproj -type f \| sort # Windows 上可用 tree /F 查看

scrapyproj/
├── scrapy.cfg                    # 部署配置，指明 settings 模块
└── scrapyproj/                   # 真正的 Python 包
    ├── __init__.py
    ├── items.py                  # Item 定义
    ├── middlewares.py            # 下载/爬虫中间件（脚手架自带，默认为空壳）
    ├── pipelines.py              # 条目管道
    ├── settings.py               # 全局配置
    └── spiders/                  # 爬虫都放这里
        └── books.py

scrapy.cfg 只有配置指向和项目名，用 scrapyd 部署时才填 [deploy]。

scrapy.cfg

[settings]
default = scrapyproj.settings

[deploy]
project = scrapyproj

items.py 里定义字段容器，继承 scrapy.Item，字段用 scrapy.Field()。

scrapyproj/items.py

import scrapy


class MovieItem(scrapy.Item):
    url = scrapy.Field()
    name = scrapy.Field()
    score = scrapy.Field()
    cover = scrapy.Field()
    regions = scrapy.Field()
    published_at = scrapy.Field()
    minute = scrapy.Field()
    director = scrapy.Field()
    actors = scrapy.Field()
    categories = scrapy.Field()
    drama_length = scrapy.Field()
    from_list = scrapy.Field()

settings.py 里决定抓取行为的关键项，本机这份就是跑通全部示例的配置。

scrapyproj/settings.py

BOT_NAME = "scrapyproj"

SPIDER_MODULES = ["scrapyproj.spiders"]
NEWSPIDER_MODULE = "scrapyproj.spiders"

ROBOTSTXT_OBEY = False

DOWNLOAD_DELAY = 0.2
CONCURRENT_REQUESTS = 8
CONCURRENT_REQUESTS_PER_DOMAIN = 4

RANDOM_USER_AGENT = True

FEED_EXPORT_ENCODING = "utf-8"

LOG_LEVEL = "DEBUG"

三个容易混的并发项：CONCURRENT_REQUESTS 是整个下载器的并发上限，CONCURRENT_REQUESTS_PER_DOMAIN 是单个域名的上限，CONCURRENT_ITEMS 只管 Pipeline 处理条目的并发数。
RANDOM_USER_AGENT = True 是 2.12 之后新加的官方开关，打开后 Scrapy 会用内置的随机 UA 中间件替换掉固定 UA 中间件，实测见「随机 UA 与反爬」。
爬虫文件用 genspider 生成，模板参数 -t 常用的是 basic 和 crawl。

1 2	scrapy genspider myspider example.com scrapy genspider probecrawl example.com -t crawl

Created spider 'myspider' using template 'basic' in module:
  scrapyproj.spiders.myspider

Created spider 'probecrawl' using template 'crawl' in module:
  scrapyproj.spiders.probecrawl

两个模板生成的代码（原样，注意 basic 模板里 parse 是 pass，crawl 模板的 rules 只有一个）：

scrapyproj/spiders/myspider.py

import scrapy


class MyspiderSpider(scrapy.Spider):
    name = "myspider"
    allowed_domains = ["example.com"]
    start_urls = ["https://example.com"]

    def parse(self, response):
        pass

scrapyproj/spiders/probecrawl.py

import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule


class ProbecrawlSpider(CrawlSpider):
    name = "probecrawl"
    allowed_domains = ["example.com"]
    start_urls = ["https://example.com"]

    rules = (Rule(LinkExtractor(allow=r"Items/"), callback="parse_item", follow=True),)

    def parse_item(self, response):
        item = {}
        return item

重名会被拦下来，但拦不住的是「名字不合规范」：本机实测 genspider BooksSpider 会顺利生成一个叫 BooksSpider.py 的模块，genspider probe-dash 也照样通过（模块名自动变成 probe_dash）。

1	scrapy genspider movie example.com

1 2	Spider 'movie' already exists in module: sc_sp_probe.spiders.movie

命名按 PEP 8 来：模块名全小写、类名大驼峰、Spider 名小写；爬虫名要能在 scrapy list 里一眼认出来。

1	scrapy list

feed
lpop
movie
movie_crawl
mvmongo
mvmysql
mvredis
mvsqlite
mvstore
rmovie
spa
spa_plain

scrapy shell

交互式调试选择器的首选，进来就有一个已抓取的 response。

1	scrapy shell "https://ssr1.scrape.center/page/1"

[s] Available Scrapy objects:
[s]   scrapy     scrapy module (contains scrapy.Request, scrapy.Selector, etc)
[s]   crawler    <scrapy.crawler.Crawler object at 0x000001A8DE291550>
[s]   item       {}
[s]   request    <GET https://ssr1.scrape.center/page/1>
[s]   response   <200 https://ssr1.scrape.center/page/1>
[s]   settings   <scrapy.settings.Settings object at 0x000001A8DE23DE50>
[s]   spider     <DefaultSpider 'default' at 0x1a8ded74d70>
[s] Useful shortcuts:
[s]   fetch(url[, redirect=True]) Fetch URL and update local objects (by default, redirects are followed)
[s]   fetch(req)                  Fetch a scrapy.Request and update local objects
[s]   shelp()           Shell help (print this help)
[s]   view(response)    View response in a browser

老教程里的 sel 在本机 2.19.0 已经不存在了，实测全局名只有这些：

1	scrapy shell "https://ssr1.scrape.center/page/1" # 进去后敲：sorted(k for k in globals() if not k.startswith('__'))

1 2	>>> sorted(k for k in globals() if not k.startswith('__')) ['crawler', 'fetch', 'item', 'request', 'response', 'scrapy', 'settings', 'shelp', 'spider', 'view']

选择器直接用 response.selector，它和当年的 sel 是同一个东西；Selector 类本身也不在全局里，要自己 import。

scrapy shell "https://ssr1.scrape.center/page/1"
# 进去后依次敲：response.selector.type /
# len(response.css("div.el-card.item")) /
# response.css("div.el-card.item::attr(class)").get() /
# len(response.xpath('//div[@class="el-card item"]')) /
# len(response.xpath('//div[contains(@class,"el-card item")]')) /
# Selector(text='<p id="x">hi</p>')

>>> response.selector.type
'html'
>>> len(response.css("div.el-card.item"))
10
>>> response.css("div.el-card.item::attr(class)").get()
'el-card item m-t is-hover-shadow'
>>> len(response.xpath('//div[@class="el-card item"]'))
0
>>> len(response.xpath('//div[contains(@class,"el-card item")]'))
10
>>> Selector(text='<p id="x">hi</p>')
NameError: name 'Selector' is not defined

⚠️ 在线站埋的第一个坑：XPath 按 class 全等取不到卡片。真实 class 是 el-card item m-t is-hover-shadow，//div[@class="el-card item"] 命中 0；要么改用 contains()，要么用 CSS div.el-card.item（CSS 走的是包含匹配，不受影响）。
shell 会强制改几个配置（实测 Overridden settings 里的这几行），所以在 shell 里能重复抓同一个 URL，而 scrapy crawl 会被去重拦掉，这点差异经常让人误判自己的爬虫。

1	scrapy shell "https://ssr1.scrape.center/page/1" # 启动日志里的 Overridden settings 就是下面这份

{'BOT_NAME': 'sc_sp_probe',
 'CONCURRENT_REQUESTS': 8,
 'CONCURRENT_REQUESTS_PER_DOMAIN': 4,
 'DOWNLOAD_DELAY': 0.2,
 'DUPEFILTER_CLASS': 'scrapy.dupefilters.BaseDupeFilter',
 'FEED_EXPORT_ENCODING': 'utf-8',
 'LOGSTATS_INTERVAL': 0,
 'LOG_LEVEL': 'INFO',
 'NEWSPIDER_MODULE': 'sc_sp_probe.spiders',
 'REMOTE_CONTROL_ENABLED': False,
 'SPIDER_MODULES': ['sc_sp_probe.spiders']}

fetch() 不返回对象，它把 shell 里的 response / request 全局换掉（实测返回 None）。

1 2	scrapy shell "https://ssr1.scrape.center/page/1" # 进去后敲：r = fetch("https://ssr1.scrape.center/page/2")；再看 r is None / response.url / 卡片数

>>> r = fetch("https://ssr1.scrape.center/page/2")
>>> r is None
True
>>> response.url
'https://ssr1.scrape.center/page/2'
>>> len(response.css("div.el-card.item"))
10

按内容类型，shell 给你的 response 子类不一样：HTML 页面是 HtmlResponse，JSON 接口是 JsonResponse。下面进 spa1 的接口，再 fetch 一个 HTML 页对比。

scrapy shell "https://spa1.scrape.center/api/movie?limit=10&offset=0"
# 进去后敲：type(response).__name__ / sorted(response.json().keys()) /
# response.json()['count'] / len(response.json()['results']) /
# response.json()['results'][0]['name'] / fetch("https://ssr1.scrape.center/page/2")

>>> type(response).__name__        # 进的是 /api/movie
'JsonResponse'
>>> sorted(response.json().keys())
['count', 'results']
>>> response.json()['count']
104
>>> len(response.json()['results'])
10
>>> response.json()['results'][0]['name']
'霸王别姬'
>>> fetch("https://ssr1.scrape.center/page/2")
>>> type(response).__name__        # fetch 换成了 HTML 页
'HtmlResponse'

不用交互也能跑

非交互（放进脚本、CI）就 --no-edit 加管道，把命令写进标准输入，末尾 exit()。

1	scrapy shell "https://ssr1.scrape.center/page/1" --no-edit < cmds.txt

但更省事的是绕开 shell，直接在普通 Python 进程里用 parsel.Selector 处理抓来的 HTML，行为与 response.selector 一致。

shell_less.py

import urllib.request

from parsel import Selector

req = urllib.request.Request("https://ssr1.scrape.center/page/1",
                             headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=10) as f:
    html = f.read().decode("utf-8")

sel = Selector(text=html)
print(sel.css("div.el-card.item a.name h2::text").getall()[:3])
print(sel.css("a.name::attr(href)").get())

1 2	['霸王别姬 - Farewell My Concubine', '这个杀手不太冷 - Léon', '肖申克的救赎 - The Shawshank Redemption'] /detail/1

另外两个同级的小工具：scrapy fetch 只打印响应体，scrapy parse 直接跑某个爬虫的回调并把条目打出来（实测用法见下一节）。

1
2
3

scrapy fetch --nolog https://ssr1.scrape.center/detail/1
scrapy fetch --nolog --headers https://ssr1.scrape.center/detail/1
scrapy parse --spider movie -d 2 -o out/parse.jl:json https://ssr1.scrape.center/page/1

注意 scrapy fetch 没有 --follow 这个选项，只有 --no-redirect；写错会直接把 usage 打到屏幕上、返回码 2。

Request 对象

参数一览（本机 inspect.signature 打出来的真实签名，默认值和类型标注都在）。

req_sig.py

import inspect
import textwrap

import scrapy

sig = str(inspect.signature(scrapy.Request.__init__))
print("Request.__init__" + textwrap.fill(
    sig, width=74, subsequent_indent=" " * 17,
    break_long_words=False, break_on_hyphens=False))

Request.__init__(self, url: 'str', callback: 'CallbackT | None' = None, method: 'str' =
                 'GET', headers: 'Mapping[str, Any] | Mapping[bytes, Any]
                 | Iterable[tuple[str | bytes, Any]] | None' = None, body:
                 'bytes | str | None' = None, cookies: 'CookiesT | None' =
                 None, meta: 'dict[str, Any] | None' = None, encoding:
                 'str' = 'utf-8', priority: 'int' = 0, dont_filter: 'bool'
                 = False, errback: 'Callable[[Failure], Any] | None' =
                 None, flags: 'list[str] | None' = None, cb_kwargs:
                 'dict[str, Any] | None' = None) -> 'None'

不传参时的默认值，全部是实测出来的：

req_defaults.py

import scrapy

r = scrapy.Request("https://ssr1.scrape.center/page/1")
for label, value in [("url", r.url), ("method", r.method), ("callback", r.callback),
                     ("errback", r.errback), ("cookies", r.cookies),
                     ("headers", dict(r.headers)), ("meta", r.meta),
                     ("encoding", r.encoding), ("priority", r.priority),
                     ("dont_filter", r.dont_filter), ("flags", r.flags),
                     ("cb_kwargs", r.cb_kwargs), ("body", r.body)]:
    print("%-14s%s" % (label, repr(value)))

url           'https://ssr1.scrape.center/page/1'
method        'GET'
callback      None
errback       None
cookies       {}
headers       {}
meta          {}
encoding      'utf-8'
priority      0
dont_filter   False
flags         []
cb_kwargs     {}
body          b''

一个都不传的 Request 也能构造，callback=None 时引擎用爬虫的 parse 兜底。

one_arg.py

import scrapy

req = scrapy.Request("https://ssr1.scrape.center/page/1")
print(req.method, req.priority, req.dont_filter, req.callback)

1	GET 0 False None

写错了会怎样

URL 必须有 scheme：空串和 javascript: 都在构造时就炸，不是等到下载阶段。

req_bad_url.py

import scrapy

for url in ("javascript:alert(1)", ""):
    try:
        scrapy.Request(url)
    except ValueError as exc:
        print("ValueError:", str(exc).rstrip())

1 2	ValueError: Missing scheme in request url: javascript:alert(1) ValueError: Missing scheme in request url:

callback 必须是可调用对象，写字符串直接 TypeError；priority 只认整数。

req_bad_arg.py

import scrapy

try:
    scrapy.Request("https://x.com", callback="parse")
except TypeError as exc:
    print("TypeError:", exc)
try:
    scrapy.Request("https://x.com", priority="high")
except TypeError as exc:
    print("TypeError:", exc)

1 2	TypeError: callback must be a callable, got str TypeError: Request priority not an integer: 'high'

method 传小写不会报错——实测会被规范成大写；body 传 str 会按 encoding 编码成 bytes；headers/cookies 的构造阶段基本不校验类型，问题留到下载阶段才暴露。

req_ctor.py

import scrapy

cases = [
    ("{'method': 'get'}", lambda: scrapy.Request("https://x.com", method="get").method),
    ("{'body': '中文'}", lambda: scrapy.Request("https://x.com", body="中文").body),
    ("{'headers': {'X': 1}}", lambda: dict(scrapy.Request("https://x.com", headers={"X": 1}).headers)),
    ("{'cookies': 'str'}", lambda: scrapy.Request("https://x.com", cookies="str").cookies),
    ("{'priority': 'high'}", lambda: scrapy.Request("https://x.com", priority="high").priority),
    ("{'cb_kwargs': 1}", lambda: scrapy.Request("https://x.com", cb_kwargs=1).cb_kwargs),
]
for label, call in cases:
    try:
        print("OK   " + label + " -> " + repr(call()))
    except TypeError as exc:
        print("FAIL " + label + " -> " + type(exc).__name__ + ": " + str(exc))

OK   {'method': 'get'} -> 'GET'
OK   {'body': '中文'} -> b'\xe4\xb8\xad\xe6\x96\x87'
OK   {'headers': {'X': 1}} -> {b'X': [b'1']}
OK   {'cookies': 'str'} -> 'str'
FAIL {'priority': 'high'} -> TypeError: Request priority not an integer: 'high'
FAIL {'cb_kwargs': 1} -> TypeError: 'int' object is not iterable

meta 是普通 dict，可以直接改；它是给中间件和回调传私活的地方，值要能被序列化，别塞爬虫实例。
replace() 返回新对象，原对象不变，用来做重试/改写很顺手。

replace_demo.py

import scrapy

r = scrapy.Request("https://ssr1.scrape.center/page/1", priority=0)
r2 = r.replace(priority=5, url="https://ssr1.scrape.center/page/2")
print(r2.priority, r2.url, r.priority)

1	5 https://ssr1.scrape.center/page/2 0

去重看的是请求指纹，同一 URL + 同 method + 同 body 的 replace() 副本指纹不变，所以「换了个对象」不等于「新请求」，还会被 RFPDupeFilter 拦。

req_fingerprint.py

import scrapy
from scrapy.dupefilters import RFPDupeFilter

r = scrapy.Request("https://ssr1.scrape.center/page/1")
copy = r.replace()
filter = RFPDupeFilter(None)
print("fingerprint 稳定:", filter.request_fingerprint(r) == filter.request_fingerprint(copy))

1	fingerprint 稳定: True

to_curl() 可以把请求导成 curl 命令，排查代理和 header 时比打印字典直观。实测 scrapy.Request("https://ssr1.scrape.center/page/1", headers={"User-Agent": "X"}).to_curl() → curl -X GET https://ssr1.scrape.center/page/1 -H 'User-Agent: X'。

Response 与 Selector

response 上常用属性的实测取值（https://ssr1.scrape.center/detail/1 的响应）：

resp_attrs_demo.py

import scrapy
from scrapy.crawler import CrawlerProcess
from scrapy.http import TextResponse


class RespSpider(scrapy.Spider):
    name = "resp"
    custom_settings = {"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0",
                       "ROBOTSTXT_OBEY": False}
    start_urls = ["https://ssr1.scrape.center/detail/1"]

    def parse(self, response):
        req = response.request
        print("status:", response.status)
        print("url:", response.url)
        print("encoding:", response.encoding)
        print("body type:", type(response.body), len(response.body))
        print("text[:60]:", repr(response.text[:60]))
        print("headers['Content-Type']:", response.headers["Content-Type"])
        print("headers.getlist('Content-Type'):", response.headers.getlist("Content-Type"))
        print("request is req:", response.request is req)
        print("flags:", response.flags)
        print("meta['download_slot']:", repr(response.meta.get("download_slot")),
              "| meta['download_timeout']:", response.meta.get("download_timeout"))
        print("ip_address:", response.ip_address,
              "| protocol:", response.protocol,
              "| certificate:", response.certificate)
        print("cls:", type(response).__name__,
              "| instanceof TextResponse:", isinstance(response, TextResponse))


process = CrawlerProcess()
process.crawl(RespSpider)
process.start()

status: 200
url: https://ssr1.scrape.center/detail/1
encoding: utf-8
body type: <class 'bytes'> 398076
text[:60]: '<html lang="en">\n<head>\n  \n  <meta charset="utf-8">\n  <meta '
headers['Content-Type']: b'text/html; charset=utf-8'
headers.getlist('Content-Type'): [b'text/html; charset=utf-8']
request is req: True
flags: []
meta['download_slot']: 'ssr1.scrape.center' | meta['download_timeout']: 180.0
ip_address: 119.188.175.109 | protocol: HTTP/1.1 | certificate: <Certificate Subject=b'scrape.center' Issuer=b'YE2'>
cls: HtmlResponse | instanceof TextResponse: True

想拿「字符集」不要写 response.charset，实测没有这个属性，只有 encoding。

resp_charset_demo.py

import scrapy

resp = scrapy.http.HtmlResponse("https://ssr1.scrape.center/detail/1",
                                body=b"<html></html>", encoding="utf-8")
try:
    resp.charset
except AttributeError as exc:
    print(type(exc).__name__ + ":", exc)

1	AttributeError: 'HtmlResponse' object has no attribute 'charset'

response.status 只是数字，但 4xx/5xx 默认根本进不了回调：HttpErrorMiddleware 会先把它们丢掉（见「重试与超时」）。要让回调收到，得在 spider 上配 handle_httpstatus_list 或 HTTPERROR_ALLOWED_CODES。实测把 /page/2/（带尾斜杠，404）放进 HTTPERROR_ALLOWED_CODES = [404] 后回调才收到：

http404_demo.py

import scrapy
from scrapy.crawler import CrawlerProcess


class Http404Spider(scrapy.Spider):
    name = "http404"
    custom_settings = {"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0",
                       "ROBOTSTXT_OBEY": False, "HTTPERROR_ALLOWED_CODES": [404]}
    start_urls = ["https://ssr1.scrape.center/page/2/"]

    def parse(self, response):
        print("回调收到 404 status:", response.status)


process = CrawlerProcess()
process.crawl(Http404Spider)
process.start()

1	回调收到 404 status: 404

反过来，不带尾斜杠的 /page/2 才是 200；不配 allowed 时访问 /page/11（500）会被直接丢：Ignoring response <500 https://ssr1.scrape.center/page/11>: HTTP status code is not handled or not allowed。
response.json() 会缓存，多次调用返回同一个对象（在 spa1 /api/movie 的 JsonResponse 上）；在 ssr1 的 HTML 上调用就是 JSONDecodeError。

resp_json_demo.py

import json

import scrapy
from scrapy.crawler import CrawlerProcess


class JsonSpider(scrapy.Spider):
    name = "json"
    custom_settings = {"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0",
                       "ROBOTSTXT_OBEY": False}
    start_urls = ["https://spa1.scrape.center/api/movie?limit=10&offset=0"]

    def parse(self, response):
        print("json() keys:", sorted(response.json().keys()))
        print("cached same object:", response.json() is response.json())
        yield scrapy.Request("https://ssr1.scrape.center/detail/1", callback=self.on_html)

    def on_html(self, response):
        try:
            response.json()
        except json.JSONDecodeError as exc:
            print("json() on html:", type(exc).__name__, exc)


process = CrawlerProcess()
process.crawl(JsonSpider)
process.start()

1
2
3

json() keys: ['count', 'results']
cached same object: True
json() on html: JSONDecodeError Expecting value: line 1 column 1 (char 0)

Selector 取值

一个 response 就是一个 Selector，css() 和 xpath() 都返回 SelectorList。

selector_demo.py

import urllib.request

from parsel import Selector

req = urllib.request.Request("https://ssr1.scrape.center/detail/1",
                             headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=15) as f:
    page = f.read().decode("utf-8")

sel = Selector(text=page)
print(sel.css("h2::text").getall())
print(sel.css("button.category span::text").getall())
print(sel.css(".director p.name::text").get())
print(sel.css("p[class*=score]").re_first(r"\d+\.\d"))

['霸王别姬 - Farewell My Concubine', '导演', '演员', '剧照']
['剧情', '爱情']
陈凯歌
9.5

⚠️ 详情页里 p[class*=score] 取到的评分带一串换行缩进（\n 9.5），所以要么 .strip()，要么像这里用 re_first 直接抽数字。
取值的四个入口：get() 拿第一个（没有就 None，不报错）、getall() 拿全部列表、re() 直接套正则、re_first() 拿第一个匹配。

selector_value_demo.py

import urllib.request

from parsel import Selector

req = urllib.request.Request("https://ssr1.scrape.center/detail/1",
                             headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=20) as f:
    sel = Selector(text=f.read().decode("utf-8"))

score = sel.css("p[class*=score]")
print("getall on empty:", sel.css("div.nope::text").getall(),
      "get:", sel.css("div.nope::text").get())
print("selector.re():", score.re(r"\d+\.\d"))
print("selector.re_first():", score.re_first(r"\d+\.\d"))

1
2
3

getall on empty: [] get: None
selector.re(): ['9.5']
selector.re_first(): 9.5

::text / ::attr(name) 是伪元素；::text 会把标签之间的空白也算成文本，ssr1 一张卡片里光 ::text 就有 55 段（大半是换行缩进）：

pseudo_demo.py

import urllib.request

from parsel import Selector

req = urllib.request.Request("https://ssr1.scrape.center/page/1",
                             headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=20) as f:
    sel = Selector(text=f.read().decode("utf-8"))

card = sel.css("div.el-card.item")[0]
print("css '::text' on one card ->", len(card.css("::text").getall()), "段")
print("css 'a.name::attr(href)' ->", card.css("a.name::attr(href)").get())
print("css 'div.el-card.item::attr(class)' ->", sel.css("div.el-card.item::attr(class)").get())

1
2
3

css '::text' on one card -> 55 段
css 'a.name::attr(href)' -> /detail/1
css 'div.el-card.item::attr(class)' -> el-card item m-t is-hover-shadow

XPath 与 CSS 可以互相嵌套，注意 XPath 里 / 是绝对路径、// 从当前节点往下找、.// 才限定当前节点往下。

xpath_axis_demo.py

import urllib.request

from parsel import Selector

req = urllib.request.Request("https://ssr1.scrape.center/page/1",
                             headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=20) as f:
    sel = Selector(text=f.read().decode("utf-8"))

card = sel.css("div.el-card.item")[0]
title = card.xpath('.//a[@class="name"]//text()').re_first(r"\S.*")
print('限定当前卡片 .//a[@class="name"]//text():', title)
print('整页绝对 //text()[contains(.,"共")]:', sel.xpath('//text()[contains(.,"共")]').get())

1 2	限定当前卡片 .//a[@class="name"]//text(): 霸王别姬 - Farewell My Concubine 整页绝对 //text()[contains(.,"共")]: 共 104 条

老代码里的 extract() / extract_first() 仍然可用（等价于 getall() / get()）；不过 SelectorList.first 在本机 parsel 1.12.1 上已经没了。

slist_first_demo.py

from parsel import Selector

sel = Selector(text="<ul><li>a</li><li>b</li></ul>")
try:
    getattr(sel.css("li"), "first")
except AttributeError as exc:
    print(type(exc).__name__ + ":", exc)

1	AttributeError: 'SelectorList' object has no attribute 'first'

要第一个节点用下标 [0]（配合 if not node: 判空）或者 .get()，要遍历就直接 for li in sel.css("li.item")，每个子节点自己还能再 .css()。

Item 与 ItemLoader

Item 是个带字段约束的 dict：能 dict()、能 in、keys()，但读没赋过值的字段和写不存在的字段都抛 KeyError。

item_demo.py

from scrapy import Field, Item


class MovieItem(Item):
    name = Field()
    score = Field()
    regions = Field()


print("type mro:", [c.__name__ for c in MovieItem.__mro__])
b = MovieItem()
b["name"] = "霸王别姬 - Farewell My Concubine"
b["score"] = 9.5
print("dict(b):", dict(b), "| len:", len(b), "| in:", "name" in b)
print("keys:", list(b.keys()), "| values:", list(b.values()))
try:
    b["nope"] = 1
except KeyError as exc:
    print("undefined field: KeyError", str(exc))
try:
    print(b["regions"])
except KeyError as exc:
    print("read undefined: KeyError", exc.args)

type mro: ['MovieItem', 'Item', 'MutableMapping', 'Mapping', 'Collection', 'Sized', 'Iterable', 'Container', 'object_ref', 'object']
dict(b): {'name': '霸王别姬 - Farewell My Concubine', 'score': 9.5} | len: 2 | in: True
keys: ['name', 'score'] | values: ['霸王别姬 - Farewell My Concubine', 9.5]
undefined field: KeyError 'MovieItem does not support field: nope'
read undefined: KeyError ('regions',)

报错细节：item["nope"] = 1 抛的是带说明字符串的 KeyError，print(item["nope"]) 抛的是空 tuple 的 KeyError，看到 KeyError: ('nope',) 就知道是「字段合法但没赋值」，看到 does not support field 就是「字段名写错了」。
scrapy.Field() 本身只是 dict 的子类，字段元信息全在那个 dict 里。

field_mro_demo.py

from scrapy import Field, Item

print("Field:", Field, "mro:", [c.__name__ for c in Field.__mro__])
print("Item mro:", [c.__name__ for c in Item.__mro__])

1 2	Field: <class 'scrapy.item.Field'> mro: ['Field', 'dict', 'object'] Item mro: ['Item', 'MutableMapping', 'Mapping', 'Collection', 'Sized', 'Iterable', 'Container', 'object_ref', 'object']

scrapy.Field(default=42) 这种默认值在本机不会被 Item 兑现：新建的 Item 是空的，读默认字段照样 KeyError，ItemLoader().load_item() 也是空的。

field_default_demo.py

from scrapy import Field, Item
from scrapy.loader import ItemLoader


class WithDefault(Item):
    a = Field()
    b = Field(default=42)


w = WithDefault()
print("fresh item dict:", dict(w))
print("fields:", dict(WithDefault.fields))
try:
    w["b"]
except KeyError as exc:
    print("  w['b'] -> KeyError:", exc.args)
print("loader honors defaults:", dict(ItemLoader(item=WithDefault()).load_item()))

fresh item dict: {}
fields: {'a': {}, 'b': {'default': 42}}
  w['b'] -> KeyError: ('b',)
loader honors defaults: {}

也别去调 load_item(defaults={...})，itemloaders 1.5.0 的签名里没有这个参数。

load_defaults_demo.py

from scrapy import Field, Item
from scrapy.loader import ItemLoader


class One(Item):
    a = Field()


try:
    ItemLoader(item=One()).load_item(defaults={"a": 1})
except TypeError as exc:
    print("TypeError:", exc)

1	TypeError: ItemLoader.load_item() got an unexpected keyword argument 'defaults'

结论：给字段兜底要么在爬虫里显式 item["x"] = ... or "-"，要么在 Pipeline 里补；别指望 Field(default=...)。
ItemAdapter 是屏蔽 Item 容器类型的官方入口，Item、dataclass、attrs、普通 dict 都能包，写通用 Pipeline 时只认它。

adapter_demo.py

import dataclasses

from itemadapter import ItemAdapter


@dataclasses.dataclass
class DCItem:
    title: str
    price: float = 0.0


print(dict(ItemAdapter(DCItem("肖申克的救赎"))))
print(dict(ItemAdapter({"k": "v"})))

1 2	{'title': '肖申克的救赎', 'price': 0.0} {'k': 'v'}

ItemLoader 加工链

Loader 把「取」和「加工」分开：add_css / add_xpath / add_value 收原始值，load_item() 一次性走完输入处理器和输出处理器。

loader_demo.py

import urllib.request

from itemloaders.processors import Join, MapCompose, TakeFirst
from parsel import Selector
from scrapy import Field, Item
from scrapy.loader import ItemLoader


class Movie(Item):
    name = Field()
    score = Field()
    regions = Field()
    categories = Field()


def to_score(value):
    return float(value)


class MovieLoader(ItemLoader):
    default_item_class = Movie
    default_input_processor = MapCompose(str.strip, default=None)
    name_out = TakeFirst()
    score_in = MapCompose(str.strip, to_score, default=None)
    score_out = TakeFirst()
    regions_out = Join("、")
    categories_out = Join("、")


url = "https://ssr1.scrape.center/detail/1"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=20) as f:
    page = f.read().decode("utf-8")

loader = MovieLoader(selector=Selector(text=page))
loader.add_css("name", "h2::text")
loader.add_css("score", "p[class*=score]::text")
loader.add_value("regions", ["中国内地", "中国香港"])
loader.add_css("categories", "button.category span::text")
print(loader.load_item())

{'categories': '剧情、爱情',
 'name': '霸王别姬 - Farewell My Concubine',
 'regions': '中国内地、中国香港',
 'score': 9.5}

处理器可以从 itemloaders.processors 拿到全部：

processors_list_demo.py

import itemloaders.processors as processors

print(sorted(n for n in dir(processors) if not n.startswith("_")
             and n not in {"TYPE_CHECKING", "annotations"}))

1	['Any', 'ChainMap', 'Compose', 'Identity', 'Join', 'MapCompose', 'SelectJmes', 'TakeFirst', 'arg_to_iter', 'wrap_loader_context']

三种 add 方法都会累积值，输入处理器逐个作用；re= 参数在处理器之前先做一次正则抽取。

add_accumulate_demo.py

from itemloaders.processors import Join, TakeFirst
from parsel import Selector
from scrapy import Field, Item
from scrapy.loader import ItemLoader


class Cat(Item):
    name = Field()
    regions = Field()


class CatLoader(ItemLoader):
    default_item_class = Cat
    name_out = TakeFirst()
    regions_out = Join("、")


ldr = CatLoader(item=Cat())
ldr.add_value("regions", "剧情")
ldr.add_value("regions", "爱情")
print("add_value 多次累积 + Join：" + repr(ldr.get_output_value("regions")))
ldr2 = CatLoader(selector=Selector(text="<h2>霸王别姬</h2>"))
ldr2.add_css("name", "h2::text", re=r"(.{4})")
print("add_css(..., re=r'(.{4})')：" + str({"name": ldr2.get_collected_values("name")}))

1 2	add_value 多次累积 + Join：'剧情、爱情' add_css(..., re=r'(.{4})')：{'name': ['霸王别姬']}

坑集中在空值和类型上：

目标字段不在 Item 里，直接 KeyError，报错原文就是字段名。

unknown_field_demo.py

from scrapy import Field, Item
from scrapy.loader import ItemLoader


class One(Item):
    name = Field()


ldr = ItemLoader(item=One())
try:
    ldr.add_value("nope", 1)
except KeyError as exc:
    print("unknown field: KeyError", repr(str(exc)))

1	unknown field: KeyError "'nope'"

MapCompose(str.strip) 遇到 None 会炸穿整条处理链，抛两层 ValueError，中间还夹着 TypeError。

map_compose_none_demo.py

from itemloaders.processors import MapCompose
from scrapy import Field, Item
from scrapy.loader import ItemLoader


class NameOnly(Item):
    name = Field()


class BadLoader(ItemLoader):
    default_item_class = NameOnly
    name_in = MapCompose(str.strip)


try:
    BadLoader().add_value("name", ["  x  ", None])
except ValueError as exc:
    print("ValueError:", exc)

ValueError: Error with input processor MapCompose: field='name' value=['  x  ', None] error='ValueError: Error in MapCompose with <method 'strip' of 'str' objects> value=['  x  ', None] error='TypeError: descriptor 'strip' for 'str' objects doesn't apply to a 'NoneType' object''

MapCompose 默认会把 None 传下去，加个 default=None 才吃：MapCompose(str.strip, default=None)；或者用 SelectJmes／Compose 提前把空值滤掉。

safe_compose.py

from itemloaders.processors import MapCompose

clean = MapCompose(str.strip, default=None)
print(clean("  霸王别姬  "))

['霸王别姬']

Loader 可以复用（load_item() 之后接着 add_value 再 load_item），也可以套嵌做分项结构；get_output_value("field") 能单独看一个字段的加工结果，调试处理链比打印整个 item 方便。

loader_reuse_demo.py

from itemloaders.processors import Join, TakeFirst
from parsel import Selector
from scrapy import Field, Item
from scrapy.loader import ItemLoader


class Movie(Item):
    name = Field()
    score = Field()
    regions = Field()


class MovieLoader(ItemLoader):
    default_item_class = Movie
    name_out = TakeFirst()
    score_out = TakeFirst()
    regions_out = Join("、")


ldr = MovieLoader(selector=Selector(text="<h2>霸王别姬</h2>"))
ldr.add_css("name", "h2::text")
first = dict(ldr.load_item())
ldr.add_value("score", 9.5)
second = dict(ldr.load_item())
ldr.add_value("regions", ["中国内地", "中国香港"])
print("loader reuse after load:", first, "->", second)
print("get_output_value('regions'):", repr(ldr.get_output_value("regions")))

1 2	loader reuse after load: {'name': '霸王别姬'} -> {'name': '霸王别姬', 'score': 9.5} get_output_value('regions'): '中国内地、中国香港'

回调链与 yield

一个 spider 从头到尾被调用了哪些方法，实测跑 ssr1 列表页打点：

flow_demo.py

import scrapy
from scrapy.crawler import CrawlerProcess

BASE = "https://ssr1.scrape.center"


class FlowSpider(scrapy.Spider):
    name = "flow"
    custom_settings = {"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0"}
    start_urls = [BASE + "/page/1"]

    def parse(self, response):
        m = response.meta
        # DepthMiddleware 会在这个回调返回后把 depth 初始化成 0
        print("parse meta: download_slot=%r download_timeout=%s "
              "download_latency=%s depth=%s"
              % (m.get("download_slot"), m.get("download_timeout"),
                 round(m.get("download_latency", 0), 3), m.get("depth", 0)))


process = CrawlerProcess()
process.crawl(FlowSpider)
process.start()

1	parse meta: download_slot='ssr1.scrape.center' download_timeout=180.0 download_latency=0.385 depth=0

__init__ 在最前，from_crawler 紧随其后 —— 需要拿 crawler（读 settings、接信号）就重写后者，别在 __init__ 里取。上面那行「调用顺序」是把 __init__ / from_crawler / start / parse / closed 各打一个点串起来的顺序说明，download_latency 每次略有出入。
⚠️ 2.19 里起始请求只能用 async def start()（异步生成器）或 start_urls。实测四种写法（各自单独跑一个只抓 /page/1 的 spider）：

start_forms_demo.py

# -*- coding: utf-8 -*-
import subprocess
import sys
import textwrap

FORMS = {
    "async def start()": """
import scrapy
from scrapy.crawler import CrawlerProcess


class S(scrapy.Spider):
    name = "s"

    async def start(self):
        yield scrapy.Request("https://ssr1.scrape.center/page/1")

    def parse(self, response):
        yield {"ok": 1}

    def closed(self, reason):
        s = self.crawler.stats.get_stats()
        print("结果:", reason, "item_scraped_count =", s.get("item_scraped_count"))


process = CrawlerProcess({"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0"})
process.crawl(S)
process.start()
""",
    "def start()（普通生成器）": """
import scrapy
from scrapy.crawler import CrawlerProcess


class S(scrapy.Spider):
    name = "s"

    def start(self):
        yield scrapy.Request("https://ssr1.scrape.center/page/1")

    def parse(self, response):
        yield {"ok": 1}


process = CrawlerProcess({"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0"})
process.crawl(S)
process.start()
""",
    "只写 start_urls": """
import scrapy
from scrapy.crawler import CrawlerProcess


class S(scrapy.Spider):
    name = "s"
    start_urls = ["https://ssr1.scrape.center/page/1"]

    def parse(self, response):
        yield {"ok": 1}

    def closed(self, reason):
        s = self.crawler.stats.get_stats()
        print("结果:", reason, "item_scraped_count =", s.get("item_scraped_count"))


process = CrawlerProcess({"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0"})
process.crawl(S)
process.start()
""",
    "只写 start_requests()": """
import scrapy
from scrapy.crawler import CrawlerProcess


class S(scrapy.Spider):
    name = "s"

    def start_requests(self):
        return [scrapy.Request("https://ssr1.scrape.center/page/1")]

    def parse(self, response):
        yield {"ok": 1}

    def closed(self, reason):
        s = self.crawler.stats.get_stats()
        print("结果:", reason, "item_scraped_count =", s.get("item_scraped_count"))


process = CrawlerProcess({"LOG_LEVEL": "ERROR", "USER_AGENT": "Mozilla/5.0"})
process.crawl(S)
process.start()
""",
}

for label, code in FORMS.items():
    p = subprocess.run([sys.executable, "-X", "utf8", "-c", textwrap.dedent(code)],
                       capture_output=True, text=True, encoding="utf-8", timeout=180)
    tail = [l for l in (p.stdout + p.stderr).splitlines() if l.strip()][-1:]
    print("%-24s -> %s" % (label, tail[0][:110] if tail else "无输出"))

async def start()        -> 结果: finished item_scraped_count = 1
def start()（普通生成器）       -> TypeError: __main__.S.start() must be an asynchronous generator, i.e. an async def method with yield statements
只写 start_urls          -> 结果: finished item_scraped_count = 1
只写 start_requests()      -> 结果: finished item_scraped_count = None

旧教程里 def start() 生成器直接报 TypeError，start_requests() 被引擎静默忽略、空跑结束。

⚠️ request.depth 在 2.19 已经没了，实测 AttributeError: 'Request' object has no attribute 'depth'；深度现在在 response.meta["depth"] 里（上面那行 meta 就能看到）。同理 response.request.referrer 也建议改读 meta。

传参给回调只用 cb_kwargs：

movie_spider.py

import scrapy

BASE = "https://ssr1.scrape.center"


class MovieSpider(scrapy.Spider):
    name = "movie"
    allowed_domains = ["ssr1.scrape.center"]

    async def start(self):
        yield scrapy.Request(BASE + "/page/1", cb_kwargs={"page": 1}, errback=self.on_bad)

    def parse(self, response, page):
        self.logger.info("第 %s 页，depth=%s，卡片 %d 张",
                         page, response.meta["depth"], len(response.css("div.el-card.item")))
        for card in response.css("div.el-card.item"):
            yield response.follow(card.css("a.name::attr(href)").get(), callback=self.parse_detail)
        if page < 10:                     # /page/11 是 500，翻页到 10 就停
            yield response.follow(f"/page/{page + 1}", cb_kwargs={"page": page + 1})

    def parse_detail(self, response):
        yield {"name": response.css("h2::text").get(default="").strip(),
               "score": (response.css("p[class*=score]::text").get() or "").strip()}

    def on_bad(self, failure):
        self.logger.warning("errback 收到：%s", failure.value)

[movie] INFO: 第 1 页，depth=0，卡片 10 张
[movie] INFO: 第 2 页，depth=1，卡片 10 张
...
[movie] INFO: 第 10 页，depth=9，卡片 10 张
[scrapy.core.engine] INFO: Closing spider (finished)

cb_kwargs 的键必须和回调形参对得上，对不上是 TypeError；meta 也能传，但它是整个下载链路的公共字典（重试、重定向都会带着），只用来传调度相关的东西。
response.follow() 会自动补全相对链接并返回 Request，实测 Response.follow 的签名与 Request 基本一致，只是 url 可以是 Link 对象。
errback 收到的是 Failure，取异常本体用 failure.value。

去重与 priority

同一 URL 第二次请求会被过滤器吃掉，实测 dupefilter/filtered: 1；把 dont_filter=True 加上才会真发两次。
priority 必须是整数，实测传字符串直接 TypeError: Request priority not an integer: 'high'；method="get" 这种小写会被规范化成 GET。

接口型（Ajax）Spider

同一个 scrape.center 有两条路：ssr1 是服务端渲染，HTML 里就有数据；spa1 首页只有 952 字节的空壳，数据全在 /api/movie 的 JSON 里。接口页 Content-Type: application/json，Scrapy 直接给你 JsonResponse，response.json() 就能拿 dict。

接口的分页参数是 limit / offset，不是 page：?page=2 照样返回第一页（静默），得靠 offset 往后推；offset 越界时 results 变空列表但 HTTP 仍是 200，所以循环结束要判 results 是否为空，不能判状态码。

spa_spider.py

import scrapy

API = "https://spa1.scrape.center/api/movie"


class SpaSpider(scrapy.Spider):
    name = "spa"
    allowed_domains = ["spa1.scrape.center"]

    async def start(self):
        yield scrapy.Request(API + "?limit=10&offset=0", cb_kwargs={"offset": 0})

    def parse(self, response, offset):
        data = response.json()                 # JsonResponse 自带 json()
        for r in data["results"]:
            yield {"id": r["id"], "name": r["name"], "score": r["score"]}
        if data["results"]:                    # 判空而不是判状态码
            nxt = offset + len(data["results"])
            yield scrapy.Request(f"{API}?limit=10&offset={nxt}",
                                 callback=self.parse, cb_kwargs={"offset": nxt})

[spa] INFO: offset=0   取到 10 条，count=104，响应类型 JsonResponse
[spa] INFO: offset=90  取到 10 条，count=104，响应类型 JsonResponse
[spa] INFO: offset=100 取到 4  条，count=104，响应类型 JsonResponse
[spa] INFO: offset=104 取到 0  条，count=104，响应类型 JsonResponse
'item_scraped_count': 104,
'finish_reason': 'finished',

不想依赖 JsonResponse 也行：普通 scrapy.Request 拿回来后 json.loads(response.text)，两者结果一致。接口是 application/json 才会自动转成 JsonResponse，HTML 上调用 response.json() 直接 JSONDecodeError。

Request 的默认值实测：method='GET'、body=b''、cookies={}、headers={}、meta={}、encoding='utf-8'、priority=0、dont_filter=False、cb_kwargs={}、flags=[]。
调试时有个现成的转换器：request.to_curl() 实测输出 curl -X GET https://ssr1.scrape.center/page/1 -H 'User-Agent: X'，直接粘终端复现。

Pipeline

四个钩子的调用顺序是实测出来的，三个 pipeline 按 ITEM_PIPELINES 的数字排序后各打一行：

pipeline_hooks_demo.py

import scrapy
from itemadapter import ItemAdapter
from scrapy.crawler import CrawlerProcess
from scrapy.exceptions import DropItem


class ProbePipelineA:
    def __init__(self, factor=1):
        self.factor = factor

    @classmethod
    def from_crawler(cls, crawler):
        print("ProbePipelineA.from_crawler")
        return cls(crawler.settings.getint("FACTOR", 42))

    def open_spider(self, spider):
        print("A.open_spider factor=%s" % self.factor)

    def process_item(self, item, spider):
        print("A.process_item %s" % ItemAdapter(item).get("title"))
        return item

    def close_spider(self, spider):
        print("A.close_spider")


class ProbePipelineB:
    @classmethod
    def from_crawler(cls, crawler):
        print("ProbePipelineB.from_crawler")
        return cls()

    def open_spider(self, spider):
        print("B.open_spider")

    def process_item(self, item, spider):
        if "红楼" in ItemAdapter(item).get("title", ""):
            print("B.process_item -> DropItem")
            raise DropItem("内容不合格")
        print("B.process_item -> pass")
        return item

    def close_spider(self, spider):
        print("B.close_spider")


class Recorder:
    @classmethod
    def from_crawler(cls, crawler):
        print("Recorder.from_crawler")
        return cls()

    def open_spider(self, spider):
        print("Recorder.open_spider spider=hooks")

    def process_item(self, item, spider):
        print("Recorder.process_item")
        return item

    def close_spider(self, spider):
        print("Recorder.close_spider")


class HooksSpider(scrapy.Spider):
    name = "hooks"

    async def start(self):
        for title in ("肖申克的救赎", "红楼梦", "霸王别姬"):
            yield {"title": title}


process = CrawlerProcess({"LOG_LEVEL": "ERROR",
                          "ITEM_PIPELINES": {"__main__.ProbePipelineA": 100,
                                             "__main__.ProbePipelineB": 200,
                                             "__main__.Recorder": 300}})
process.crawl(HooksSpider)
process.start()

ProbePipelineA.from_crawler
ProbePipelineB.from_crawler
Recorder.from_crawler
A.open_spider factor=42
B.open_spider
Recorder.open_spider spider=hooks
A.process_item 肖申克的救赎
B.process_item -> pass
Recorder.process_item
A.process_item 红楼梦
B.process_item -> DropItem
A.process_item 霸王别姬
B.process_item -> pass
Recorder.process_item
Recorder.close_spider
B.close_spider
A.close_spider

from_crawler 全部先跑（此时 spider 还没开），然后按数字从小到大依次 open_spider；close_spider 是反序关的。
数字小的先处理 item，所以上面 B 抛 DropItem 之后 Recorder 就再也看不到那条（红楼梦那轮没有 Recorder.process_item）。

丢件有两种写法，统计里不一样，实测：

drop_kinds_demo.py

import scrapy
from itemadapter import ItemAdapter
from scrapy.crawler import CrawlerProcess
from scrapy.exceptions import DropItem


class DropPipeline:
    def process_item(self, item, spider):
        if "红楼" in ItemAdapter(item).get("title", ""):
            raise DropItem("内容不合格")
        return item


class NonePipeline:
    def process_item(self, item, spider):
        if "红楼" in ItemAdapter(item).get("title", ""):
            spider.logger.info("returned None for %s",
                               ItemAdapter(item).get("title"))
            return None
        return item


class DSpider(scrapy.Spider):
    name = "drop"

    async def start(self):
        for title in ("肖申克的救赎", "红楼梦", "霸王别姬"):
            yield {"title": title}


def dropped(pipeline):
    process = CrawlerProcess({"LOG_LEVEL": "ERROR",
                              "ITEM_PIPELINES": {"__main__." + pipeline: 100}})
    crawler = process.create_crawler(DSpider)
    crawler.crawl()
    process.start()
    return {k: v for k, v in crawler.stats.get_stats().items() if "dropped" in k}


print("raise DropItem ->", dropped("DropPipeline"))

1 2	raise DropItem -> {'item_dropped_count': 1, 'item_dropped_reasons_count/DropItem': 1} return None -> 日志里只有一行 "returned None for 红楼梦"，统计里没有任何 dropped_* 计数

要「计入统计并告诉别人为什么丢」就 DropItem；只想静默过滤就返回 None（对照 NonePipeline，把 dropped("NonePipeline") 跑一遍就是空 dict {}）。

Item 字段必填时报错很直白，实测：

required_field_demo.py

import scrapy
from itemadapter import ItemAdapter
from scrapy.crawler import CrawlerProcess


class RequiredPipeline:
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        print("keys present=%s" % sorted(adapter.keys()))
        try:
            _ = item["price"]
        except KeyError as exc:
            print("KeyError args=%r str=%s" % (exc.args, exc))
        return item


class KSpider(scrapy.Spider):
    name = "keys"

    async def start(self):
        yield {"title": "红楼梦"}


process = CrawlerProcess({"LOG_LEVEL": "ERROR",
                          "ITEM_PIPELINES": {"__main__.RequiredPipeline": 100}})
process.crawl(KSpider)
process.start()

1 2	keys present=['title'] KeyError args=('price',) str='price'

导数据别自己写循环，用 FEEDS。实测一次跑完同时出三种格式：

settings.py

FEEDS = {
    "out/feed.json": {"format": "json", "encoding": "utf-8"},
    "out/feed.jsonl": {"format": "jsonlines", "encoding": "utf-8"},
    "out/feed.csv": {"format": "csv", "encoding": "utf-8", "overwrite": True},
}
FEED_EXPORT_ENCODING = "utf-8"

[scrapy.extensions.feedexport] INFO: Stored json feed (3 items) in: out/feed.json
[scrapy.extensions.feedexport] INFO: Stored csv feed (3 items) in: out/feed.csv
'item_scraped_count': 3, 'finish_reason': 'finished', 'downloader/request_count': 4,
out/feed.csv 头一行：url,name,score,cover,regions,published_at,minute,director,actors,categories,drama_length,from_list
out/feed.csv 数据：...中国香港,1993-07-26 上映,171 分钟,陈凯歌,"张国荣,张丰毅,巩俐,...","剧情,爱情",273,True

regions / actors / categories 都是列表，CSV 里被写成带引号的 "剧情,爱情" —— 列表字段要导出干净，先在 pipeline 里 "；".join(...)。
没有 item 时默认不生成文件；要让空结果也产出文件就设 FEED_STORE_EMPTY = True（实测会写出一个只有表头的 csv）。

中间件

下载中间件的三个钩子都能在日志里看到，实测一份「随机 UA + 短路 + 拦截」的中间件：

middlewares.py

import random

UAS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36",
]


class RandomUa:
    def process_request(self, request, spider):
        request.headers["User-Agent"] = random.choice(UAS)
        return None                      # 返回 None 表示继续往下走

    def process_response(self, request, response, spider):
        return response

    def process_exception(self, request, exception, spider):
        if isinstance(exception, TimeoutError):
            request.dont_filter = True
            return request               # 返回 Request 表示重发
        return None

signal spider_opened -> hook
process_request #1 http://127.0.0.1:8010/list?page=1
RandomUa -> Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)
process_request #2 http://127.0.0.1:8010/detail/1
   短路返回 Response，不发网络请求
process_response 200 http://127.0.0.1:8010/detail/1 flags=[]
spider_mw input http://127.0.0.1:8010/detail/1
process_request #3 http://127.0.0.1:8010/detail/2
   抛 IgnoreRequest
process_exception http://127.0.0.1:8010/detail/2 IgnoreRequest: middleware blocked ...

process_request 的三种返回：None 继续、返回 Response 直接短路（不发网络请求，实测 detail/1 就是被这么截的）、返回 Request 表示换请求重排。
process_exception 返回 Request 是重发，返回 Response 是顶替，返回 None 就往下抛。
爬虫中间件（spider_mw input/output）作用在「响应进 spider」和「spider 出 item/Request」这两头，allowed_domains 的过滤就靠它。

UA 与 403

靶站 /ua 只认带 Mozilla 的 UA，用它来验证几种设置的实际效果：

1	scrapy crawl ua # 挂上面那份 RandomUa，RANDOM_USER_AGENT=True / False 各跑一遍，再跑自写中间件那次

1
2
3

RANDOM_USER_AGENT=True（默认）      -> 200，中间件随机换 UA
RANDOM_USER_AGENT=False + 不写 UA    -> Scrapy 默认 UA 是 "Scrapy/2.19.0 (+https://scrapy.org)"，403
自写 RandomUa 中间件                 -> 200，日志里能看到每次换的哪个

1
2

http://127.0.0.1:8010/ua 200 b'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ... Chrome/131.0.0.0 Safari/537.36'
http://127.0.0.1:8010/ua 200 b'Mozilla/5.0 (X11; Linux x86_64) ... Chrome/130.0.0.0 Safari/537.36'

只改 UA 不够：真实浏览器是一整套头，缺 Sec-Fetch-* 那组照样能被识别（对照《爬虫-基本入门》的实测表）。

CrawlSpider 翻页

Rule 与 LinkExtractor 的完整签名实测打印过，常用参数就这几个：

movie_crawl.py

import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule

BASE = "https://ssr1.scrape.center"


class MovieCrawlSpider(CrawlSpider):
    name = "movie_crawl"
    allowed_domains = ["ssr1.scrape.center"]
    start_urls = [BASE + "/page/1"]

    rules = [
        Rule(LinkExtractor(restrict_css="div.el-pagination", allow=r"/page/\d+"),
             callback="parse_page", follow=True, process_links="drop_over",
             cb_kwargs={"from_crawl": True}),
        Rule(LinkExtractor(restrict_css="div.el-card.item", allow=r"/detail/\d+"),
             callback="parse_detail", follow=False),
    ]

    def drop_over(self, links):
        self.logger.info("process_links 收到 %d 条: %s",
                         len(links), [link.url for link in links])
        return [link for link in links if not link.url.endswith("/page/11")]

    def parse_page(self, response, from_crawl):
        self.logger.info("parse_page %s from_crawl=%s depth=%s",
                         response.url, from_crawl, response.meta["depth"])

    def parse_detail(self, response):
        yield {"name": response.css("h2::text").get(default="").strip(),
               "score": (response.css("p[class*=score]::text").get() or "").strip()}

[movie_crawl] INFO: process_links 收到 11 条: [ .../page/1, ... /page/2, ... .../page/11 ]
[movie_crawl] INFO: parse_page https://ssr1.scrape.center/page/10 from_crawl=True depth=1
'downloader/request_count': 47,
'downloader/response_status_count/200': 47,
'item_scraped_count': 38,
'finish_reason': 'closespider_itemcount',
'elapsed_time_seconds': 12.033955099992454,

follow=True 才会继续往下一层；callback=None + follow=True 是最常见的「只跟不抽」写法。
起始页不进 callback，想让它也走同一套解析就实现 parse_start_url。
LinkExtractor 的 restrict_css / restrict_xpaths / restrict_text / allow / deny / deny_extensions / unique / canonicalize 都在签名里，allow 是正则不是字符串通配。
process_links 是防坑的关键：页码条一直给到 /page/11，但第 11 页实测是 500（Gave up retrying <GET .../page/11> (failed 3 times): 500 Internal Server Error）。这里在 drop_over 里把 /page/11 摘掉，就只跟到 10 页，全程 47 个请求全是 200。

POST、表单与 403

四种 POST 实测都能打通靶站 /api/books（返回 201）：

posts.py

import json

import scrapy
from scrapy.http import FormRequest, JsonRequest


class PostsSpider(scrapy.Spider):
    name = "posts"
    url = "http://127.0.0.1:8010/api/books"

    def start(self):
        yield scrapy.Request(self.url, method="POST", body=b"title=x",
                             headers={"Content-Type": "application/x-www-form-urlencoded"})
        yield FormRequest(self.url, formdata={"title": "x"})
        yield JsonRequest(self.url, data={"title": "x"})          # 自动带 application/json
        yield scrapy.Request(self.url, method="POST",
                             body=json.dumps({"title": "x"}).encode(),
                             headers={"Content-Type": "application/json"})

JsonRequest 是 2.19 里现成的类，dumps_kwargs 可以控制序列化；FormRequest.from_response() 用来「从页面里那个表单」构造请求，参数 formname/formid/formnumber/formxpath/formcss/clickdata/dont_click 全都在。
靶站 /submit 校验隐藏字段 token，不带对的就 403，实测日志：
1
Crawled (403) <POST http://127.0.0.1:8010/submit>

⚠️ 4xx 默认不会进 parse：HttpErrorMiddleware 会把它们丢掉，日志里是

1	scrapy crawl login # 验证码填错 -> /login 返 401；没配 handle_httpstatus_list 时只打 Ignoring

1	Ignoring response <401 http://127.0.0.1:8010/login>: HTTP status code is not handled or not allowed

要处理就在 spider 上写 handle_httpstatus_list = [401, 403]，或单个请求加 meta={"handle_httpstatus_list": [403]}。

重试与超时

实测把 /status/500 配上 RETRY_TIMES=1（默认就是 2 次重试）：

1 2	scrapy crawl net -s RETRY_TIMES=1 -s RETRY_HTTP_CODES="500" \ -a url=http://127.0.0.1:8010/status/500

1
2
3

[scrapy.downloadermiddlewares.retry] ERROR: Gave up retrying <GET http://127.0.0.1:8010/status/500>
(failed 2 times): 500 Internal Server Error
统计：retry/count: 1, retry/max_reached: 1, retry/reason_count/500 Internal Server Error: 1

超时是 DOWNLOAD_TIMEOUT（默认 180 秒，上面 meta 里能看到 download_timeout: 180.0）；对慢接口设成 1 秒就会看到 TimeoutError 进 process_exception。

下载图片与文件

ImagesPipeline 的坑全在防盗链上。不带 Referer 时实测：

1	scrapy crawl pic # DOWNLOAD_HEADERS 里不写 Referer，图片请求裸奔到靶站 /img/3.png

Crawled (403) <GET http://127.0.0.1:8010/img/3.png> (referer: None)
File (code: 403): Error downloading file from <GET http://127.0.0.1:8010/img/3.png> referred in <None>
File (unknown-error): Error downloading image from ... : download-error
scrapy.pipelines.media.FileException

加上 Referer 就全通，实测：

1	scrapy crawl pic -s DOWNLOAD_HEADERS='{"Referer": "http://127.0.0.1:8010/list"}'

1
2
3

Crawled (200) <GET http://127.0.0.1:8010/img/1.png> (referer: http://127.0.0.1:8010/list)
File (downloaded): Downloaded file from <GET http://127.0.0.1:8010/img/1.png>
                  referred in <http://127.0.0.1:8010/list>

settings.py

1
2
3

IMAGES_STORE = "media"
DOWNLOAD_HEADERS = {"Referer": "http://127.0.0.1:8010/list"}
ITEM_PIPELINES = {"scrapy.pipelines.images.ImagesPipeline": 1}

更细的粒度是在图片请求的 meta 里带 Referer（response.follow(url, meta={"referer": response.url})），这样只有图片请求带、别的请求不带。
落盘文件名是内容哈希，files 字段回填的是相对路径，实测 FilesPipeline 下 /video 得到 {'file_urls': [...], 'files': ['video/video']}。
IMAGES_THUMBS、IMAGES_URLS_FIELD、IMAGES_RESULT_FIELD、FILES_STORE 都是类属性级别的改名，不用改代码。

代理

三种粒度，实测都能让靶站的 /whoami 看到代理留下的 Via：

proxy_meta_demo.py

import scrapy

req = scrapy.Request("http://127.0.0.1:8010/whoami")
req.meta["proxy"] = "http://127.0.0.1:8011"          # 单个请求走 HTTP 代理
print("proxy ->", req.meta["proxy"])
# socks5 版：req.meta["proxy"] = "socks5://127.0.0.1:8012"（需装 PySocks）

meta 单请求代理：  request.meta["proxy"] = "http://127.0.0.1:8011"
                  -> {'via': '1.1 spider-proxy', 'xff': '127.0.0.1'}
直连（不设代理）：  -> {'via': '直连无 Via', 'xff': ''}
SOCKS5：          request.meta["proxy"] = "socks5://127.0.0.1:8012"（要装 PySocks）

想轮换就自己写中间件，在 process_request 里按队列换：

middlewares.py

class ProxyRotate:
    def __init__(self, proxies):
        self.proxies = proxies

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings.getlist("PROXY_LIST"))

    def process_request(self, request, spider):
        request.meta["proxy"] = self.proxies.pop(0)
        self.proxies.append(request.meta["proxy"])
        return None

1 2	[proxyrotate] INFO: via=1.1 spider-proxy xff=127.0.0.1 [proxyrotate] INFO: via=None xff=None # 这一条走的是没代理的那次

⚠️ 和《爬虫-代理》里同一条坑：本机 Windows 注册表的代理例外列表里有 127.0.0.1，Twisted 会读系统代理设置，测本地靶站时最好显式设 REACTOR_EVENT_LOOP 无关的 NO_PROXY，或直接换个非例外地址验证，别看到「Via 没出现」就断定代理中间件写错了。

模拟登录

靶站的登录要 sid cookie + 验证码，scrapy 里完整跑通（实测日志）：

1	scrapy crawl login # 见下面这只 login_spider.py，走 GET 登录页 -> 取验证码 -> FormRequest.from_response POST

[login] INFO: GET 登录页 status=200 cookie=b'sid=5f9e173b; Path=/; SameSite=Lax'
[login] INFO: 页面里印出来的验证码=8512
[login] INFO: POST 之后 status=200 标题='登录成功' 正文=['欢迎 lilei']
[login] INFO: 登录态检查 /readcookie -> {'cookie': 'sid=5f9e173b', 'sid': '5f9e173b'}

login_spider.py

import re

import scrapy


class LoginSpider(scrapy.Spider):
    name = "login"

    def start(self):
        return scrapy.Request("http://127.0.0.1:8010/login", callback=self.do_login)

    def do_login(self, response):
        captcha = re.search(r"真实站点不会写出来）：(\d{4})", response.text).group(1)
        return scrapy.FormRequest.from_response(
            response,
            formdata={"username": "lilei", "password": "password", "captcha": captcha},
            callback=self.after_login,
        )

    def after_login(self, response):
        self.logger.info("POST 之后 status=%s 标题=%r", response.status,
                         response.css("h1::text").get())

cookie 不用自己管：CookiesMiddleware 会把 Set-Cookie 存进 jar 并自动回带，实测 /readcookie 拿到的就是同一个 sid。
密码这里写成字面量 password 只是为了示例；真实脚本从环境变量或配置读，别把口令写进仓库。
错验证码时是 401，实测被 HttpErrorMiddleware 拦掉（Ignoring response <401 ...>），要判成败得先 handle_httpstatus_list。
真实站点的验证码要 OCR，做法见《爬虫-Requests》那一套（Tesseract / ddddocr），scrapy 里只是在 do_login 里多调一次识别函数。

存到数据库

MySQL 用 pymysql 自己写 pipeline 最省事；抓取源换成在线站后，落库这步仍打本机 MySQL，实测一次 crawl：

1	scrapy crawl mvmysql # mvstore 只抓第 1 页 10 条详情，ITEM_PIPELINES 挂下面的 MysqlPipeline

[mvmysql] INFO: mysql ready, pw_len=10
[mvmysql] INFO: mysql rows=10 total_score=94.0
[mvmysql] INFO: mysql row (1, '霸王别姬 - Farewell My Concubine', Decimal('9.5'), '中国内地、中国香港', '陈凯歌', 273)
[mvmysql] INFO: mysql row (2, '这个杀手不太冷 - Léon', Decimal('9.5'), '法国', '吕克·贝松', 234)
[mvmysql] INFO: mysql row (3, '肖申克的救赎 - The Shawshank Redemption', Decimal('9.5'), '美国', '弗兰克·德拉邦特', 322)
rows after crawl: 10

mysql_pipeline.py

import os

import pymysql
from itemadapter import ItemAdapter


class MysqlPipeline:
    DDL = """
    CREATE TABLE IF NOT EXISTS movies_probe (
      id INT PRIMARY KEY, name VARCHAR(128), score DECIMAL(4,1),
      regions VARCHAR(128), published_at VARCHAR(32), minute VARCHAR(32),
      director VARCHAR(64), categories VARCHAR(128), actors_count INT,
      drama_length INT, url VARCHAR(255)
    )
    """

    def __init__(self, conn):
        self.conn = conn

    @classmethod
    def from_crawler(cls, crawler):
        return cls(pymysql.connect(
            host="127.0.0.1", user="root",
            password=os.environ["DB_PW"],        # 口令只从环境变量来
            database="spider_probe", autocommit=True, charset="utf8mb4"))

    def open_spider(self, spider):
        with self.conn.cursor() as c:
            c.execute(self.DDL)
            c.execute("DELETE FROM movies_probe")

    def process_item(self, item, spider):
        a = ItemAdapter(item)
        mid = int(a["url"].rstrip("/").split("/")[-1])   # /detail/3 -> 3
        with self.conn.cursor() as c:
            c.execute("REPLACE INTO movies_probe "
                      "(id,name,score,regions,published_at,minute,director,"
                      "categories,actors_count,drama_length,url) "
                      "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
                      (mid, a["name"], a["score"], "、".join(a["regions"]),
                       a["published_at"], a["minute"], a["director"],
                       "、".join(a["categories"]), len(a["actors"]),
                       a["drama_length"], a["url"]))
        return item

    def close_spider(self, spider):
        self.conn.close()

score 列用 DECIMAL(4,1) 取回来是 Decimal('9.5') 而不是 float，这是 MySQL 驱动的默认行为，别当浮点比大小。
第二次跑还是 10 行（用 REPLACE INTO 按 id 去重）；直接 INSERT 就会翻倍，重复抓取的去重要自己保证。
连接放在 from_crawler / open_spider 里建，别在 __init__（那时 spider 还没有）。

SQLite 最省事，实测：

1	scrapy crawl mvsqlite # 同样用 mvstore 抓 10 条，ITEM_PIPELINES 换成 SqlitePipeline

1 2	[mvsqlite] INFO: sqlite rows=10 size=12288 sqlite sample: [('喜剧之王 - The King of Comedy', 9.5, '中国香港'), ('唐伯虎点秋香 - Flirting Scholar', 9.5, '中国香港')]

MongoDB 也跑通了（本机 9.0 单机，pymongo 4.19.0）：

1	scrapy crawl mvmongo # ITEM_PIPELINES 换成 MongoPipeline，按 url 作 _id 落库

1
2
3

mongo count=10
one doc: {"name": "狮子王 - The Lion King", "score": 9.0, "regions": ["美国"],
          "categories": ["动画", "歌舞", "冒险"], "actors": ["马修·布罗德里克", ... 共 30 个]}

列表字段（regions / categories / actors）在 Mongo 里就是数组，不用像 CSV 那样拼接；这也是「结构化数据多的场景 Mongo 更省事」的具体含义。

Redis 也可以直接落库：自写一个 pipeline，把每条 item rpush 进一个 list（口令走环境变量），实测一次 mvredis 跑完队列里正好 10 条，取回首条是 JSON 串。

1	scrapy crawl mvredis # RedisStorePipeline：open_spider 先 delete，再逐条 rpush 进 movies:items

1	[mvredis] INFO: redis llen(movies:items)=10 sample={"url": "https://ssr1.scrape.center/detail/9", "name": "楚门的世界 - The Truman Show", ...

礼貌抓取与 robots

ROBOTSTXT_OBEY=True（scrapy startproject 生成的默认值）时，有 Disallow 的站才拦得住。靶站的 Disallow: /private/ 实测被挡：

1	scrapy crawl net -s ROBOTSTXT_OBEY=True -a url=http://127.0.0.1:8010/private/

1 2	Forbidden by robots.txt: <GET http://127.0.0.1:8010/private/> 'robotstxt/forbidden': 1,

⚠️ 但在 ssr1 / spa1 上打开 ROBOTSTXT_OBEY=True 照样全抓，这才是 robots.txt 只是约定的真相：ssr1 根本没有 robots.txt（GET /robots.txt 是 404），spa1 的 robots.txt 返回 200 但 body 是 952 字节的 text/html SPA 外壳——robots.txt 拉不到规则时，protego 给一个空规则池，Scrapy 当作「没有限制、全部放行」，既不报错也不拦。

1 2	scrapy crawl movie -s ROBOTSTXT_OBEY=True -s CLOSESPIDER_PAGECOUNT=1 scrapy crawl spa -s ROBOTSTXT_OBEY=True -s CLOSESPIDER_PAGECOUNT=1

# ROBOTSTXT_OBEY=True，请求 ssr1 /page/1 前 Scrapy 先抓 robots.txt
Crawled (404) <GET https://ssr1.scrape.center/robots.txt> (referer: None)
'robotstxt/response_status_count/404': 1,          # 之后目标页照常 200，条目照抓
# spa1 同理，只是这次 robots.txt 是 200 但内容为 HTML
Crawled (200) <GET https://spa1.scrape.center/robots.txt> (referer: None)
'robotstxt/response_status_count/200': 1,

并发与限速：在 ssr1 全站（10 个列表页 + 100 个详情页 = 110 个请求、item_scraped_count=100）上各跑一次，实测对照：

scrapy crawl movie -s CONCURRENT_REQUESTS=4  -s CONCURRENT_REQUESTS_PER_DOMAIN=4 -s DOWNLOAD_DELAY=0.5 -s ROBOTSTXT_OBEY=False
scrapy crawl movie -s CONCURRENT_REQUESTS=16 -s CONCURRENT_REQUESTS_PER_DOMAIN=16 -s DOWNLOAD_DELAY=0   -s ROBOTSTXT_OBEY=False
scrapy crawl movie -s CONCURRENT_REQUESTS=8  -s DOWNLOAD_DELAY=0 -s AUTOTHROTTLE_ENABLED=True -s AUTOTHROTTLE_TARGET_CONCURRENCY=8 -s ROBOTSTXT_OBEY=False
scrapy crawl movie -s CONCURRENT_REQUESTS=4  -s CONCURRENT_REQUESTS_PER_DOMAIN=4 -s DOWNLOAD_DELAY=0.2 -s ROBOTSTXT_OBEY=False

档位        关键配置                                       耗时      finish_reason  请求  条目
A 保守礼貌  REQUESTS=4 PER_DOMAIN=4 DELAY=0.5            67.69s   finished        110   100
B 高并发    REQUESTS=16 PER_DOMAIN=16 DELAY=0            13.22s   finished        110   100
C 自动节流  REQUESTS=8 DELAY=0 AUTOTHROTTLE=1            20.54s   finished        110   100
主线默认    REQUESTS=4 PER_DOMAIN=4 DELAY=0.2            36.03s   finished        110   100

四档 finish_reason 全是 finished、response_status_count/200 都是 110：这个站对并发不敏感、不校验头也不限速，耗时主要由本机带宽和它自己的响应时间决定。
CONCURRENT_ITEMS 只管 Pipeline 消费条目的并发，跟下载无关：同一批条目把它从 1 调到 100，elapsed 是 8.42s vs 7.89s，几乎没差 —— 落库/写文件这点活根本排不满。

HTTP 缓存中间件（HttpCacheMiddleware）实测第二次跑同一批 URL 时命中磁盘缓存，downloader/response_count 明显下降；调试期想看到真实请求就 --set HTTPCACHE_ENABLED=False。
CLOSESPIDER_ITEMCOUNT、CLOSESPIDER_PAGECOUNT、CLOSESPIDER_TIMEOUT 用来「抓够就停」，实测 CLOSESPIDER_ITEMCOUNT=2 时 finish_reason 变成 closespider_itemcount。

日志与统计

一次完整运行的统计字典（MySQL 那次，实测原样）：

1	scrapy crawl movie 2>&1 \| grep -A 30 "Dumping Scrapy stats"

STATS: {'downloader/request_bytes': 30949,
 'downloader/request_count': 110,
 'downloader/request_method_count/GET': 110,
 'downloader/response_bytes': 18274043,
 'downloader/response_count': 110,
 'downloader/response_status_count/200': 110,
 'elapsed_time_seconds': 36.034514200058766,
 'finish_reason': 'finished',
 'item_scraped_count': 100,
 'items_per_minute': 166.66666666666669,
 'log_count/INFO': 14,
 'request_depth_max': 10,
 'response_received_count': 110,
 'scheduler/dequeued': 110,
 'scheduler/enqueued': 110,
 'start_time': datetime.datetime(2026, 11, 6, 20, 59, 59, 567140, tzinfo=datetime.timezone.utc)}

上面是 scrapy crawl movie 抓 ssr1 全站（10 页 + 100 详情 = 110 请求、100 条目）的原样统计。
看这几项就够定位问题：response_status_count/4xx 涨 → 被反爬或路径错；dupefilter/filtered 涨 → 重复请求；retry/max_reached 非 0 → 站点不稳或代理不通；item_scraped_count 为 0 但 response_count 正常 → 选择器没命中。

日志等级：LOG_LEVEL 或命令行 -L INFO，LOG_FILE 落盘。实测 -L INFO 时统计仍然会打（Dumping Scrapy stats 本身是 INFO），但 DEBUG 的逐条 Crawled (...) 就没了 —— 排查选择器问题时别把等级调太高。

分布式（scrapy-redis）

组件是齐的：scrapy-redis 0.9.1 + redis 8.1.0，配置四件套：

scrapy-redis 分布式

settings.py

SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://:password@127.0.0.1:6379/9"     # 口令走环境变量，别写死
SCHEDULER_PERSIST = True
ITEM_PIPELINES = {"scrapy_redis.pipelines.RedisPipeline": 300}

spider 侧还要把种子放进 Redis，并用 redis_key 指定队列名：

rmovie.py

from scrapy_redis.spiders import RedisSpider


class RMovieSpider(RedisSpider):
    name = "rmovie"
    allowed_domains = ["ssr1.scrape.center"]
    redis_key = "movies:urls"          # 不是 start_urls，种子从 Redis 队列取

复测结论（和旧稿相反）：本机 scrapy-redis 0.9.1 + Scrapy 2.19.0 这次能跑。调度器、去重器正常加载，种子被消费，条目写进了 Redis —— 跑 rmovie 前 llen(movies:urls)=1，跑完变 0，scheduler/dequeued/redis=1，rmovie:items 里也躺了一条：

1
2
3

python -c "import os,redis;r=redis.Redis.from_url('redis://:'+os.environ['DB_PW']+'@127.0.0.1:6379/9');r.delete('movies:urls','rmovie:items');print('跑前队列长度 =',r.rpush('movies:urls','https://ssr1.scrape.center/page/1'))"
scrapy crawl rmovie -s CLOSESPIDER_TIMEOUT=40
python -c "import os,redis;r=redis.Redis.from_url('redis://:'+os.environ['DB_PW']+'@127.0.0.1:6379/9');print('跑后队列长度 =',r.llen('movies:urls'),'| rmovie:items =',r.llen('rmovie:items'),r.lrange('rmovie:items',0,0))"

跑前队列长度 = 1
'finish_reason': 'closespider_timeout',          # RedisSpider 会一直空等下一批种子，靠超时收尾
'item_scraped_count': 1,
'downloader/request_count': 1,
'scheduler/dequeued/redis': 1,
'scheduler/enqueued/redis': 1,
跑后队列长度 = 0 | rmovie:items = 1
 [b'{"url": "https://ssr1.scrape.center/page/1", "cards": 10}']

代价是启动会刷一堆 ScrapyDeprecationWarning（RedisPipeline.process_item() 还带 spider 参数、StatsCollector.inc_value() 的 spider 参数、pipeline 返回 Deferred 等），说明它只是「还能跑」，适配层并没跟上 2.19。

想要更干净、可控的分布式，自己实现一个从 Redis 取种子的 spider 更稳 —— 在 start() 里 lpop 就行：

lpop_spider.py

import os

import redis
import scrapy


class LpopSpider(scrapy.Spider):
    name = "lpop"
    allowed_domains = ["ssr1.scrape.center"]

    def __init__(self, *a, **kw):
        super().__init__(*a, **kw)
        self.r = redis.Redis(host="127.0.0.1", port=6379, db=9,
                             password=os.environ["DB_PW"])

    async def start(self):
        while True:
            url = self.r.lpop("movies:urls")
            if url is None:
                break
            yield scrapy.Request(url.decode())

    def parse(self, response):
        yield {"url": response.url,
               "cards": len(response.css("div.el-card.item"))}

[lpop] INFO: 从队列取了 3 个种子
'item_scraped_count': 3,
'finish_reason': 'finished',
'elapsed_time_seconds': 0.7628531999653205,
跑完 llen(movies:urls) = 0

这就是「共享队列 + 多进程」的最小可用形态：每个进程都跑同一个 spider，队列谁抢到算谁的，去重要自己用 Set 记。

常见报错对照

报错原文	真正原因
`ModuleNotFoundError: No module named 'scrapy.items'`	模块名是 `scrapy.item`（单数），或直接 `from scrapy import Item, Field`
`AttributeError: 'Request' object has no attribute 'depth'`	2.19 把深度挪进 `response.meta["depth"]`
`TypeError: ...start() must be an asynchronous generator`	起始请求要用 `async def start()` 或 `start_urls`，普通 `def start` 和 `start_requests()` 都不认
`AttributeError: 'SelectorList' object has no attribute 'first'`	取第一个用 `.get()`，全部用 `.getall()`
`TypeError: Request priority not an integer: 'high'`	`priority` 只能是整数
`TypeError: callback must be a callable, got str`	回调要传函数（或 spider 上的方法名，但不能是裸字符串变量）
`Ignoring response <401 ...>: HTTP status code is not handled or not allowed`	4xx 被 `HttpErrorMiddleware` 丢了，加 `handle_httpstatus_list`
`Gave up retrying <GET ...> (failed 2 times): 500 Internal Server Error`	重试次数用完，看 `RETRY_TIMES` / `RETRY_HTTP_CODES`
`Forbidden by robots.txt: <GET ...>`	`ROBOTSTXT_OBEY=True` 在起作用
`KeyError: 'price'`（pipeline 里）	Item 少了必填字段，选择器没命中
`NotConfigured: Pipeline is not enabled`	`ITEM_PIPELINES` 里没注册，或数字写错
`Spider 'books' already defined`	同名的 spider 有两个文件，删掉或改名
`FileException`（图片 pipeline）	媒体下载失败，多半是防盗链，见「下载图片与文件」

这篇讲什么

装与版本

老写法的报错原文

工程骨架

scrapy shell

不用交互也能跑

Request 对象

写错了会怎样

Response 与 Selector

Selector 取值

Item 与 ItemLoader

ItemLoader 加工链

回调链与 yield

去重与 priority

接口型（Ajax）Spider

Pipeline

中间件

UA 与 403

CrawlSpider 翻页

POST、表单与 403

重试与超时

下载图片与文件

代理

模拟登录

存到数据库

礼貌抓取与 robots

日志与统计

分布式（scrapy-redis）

常见报错对照

参考