爬虫-Feapder

发表于 更新于 分类于 本文字数： 148k 阅读时长 ≈ 4:30

这篇讲什么

  • feapder 1.9.3 在 Python 3.14.10 上逐条实测：四种爬虫（AirSpider / Spider / TaskSpider / BatchSpider）、命令行脚手架、Item 与数据管道、去重、代理、浏览器渲染、失败重试与报警，抓取对象统一换成公开练习站 ssr1.scrape.center 与 spa1.scrape.center，输出全部逐字贴出。
  • 装它必须单独开虚拟环境：它锁 redis<4.0.0，pip install --dry-run 显示会把已有的 redis 8.1.0 悄悄换成 3.5.3，一个 warning 都不给；Python 3.12 起标准库删掉 distutils，不装 setuptools 连 import feapder 都过不去。
  • 断点续爬和任务防丢做了破坏性实验：抓到第 8 条时 os._exit 硬退出，redis 里同时留下 score=300（待做）和 score=当前时间+600（取出中）两类任务，重启后打印「检查到有待做任务 9 条，不重下发新任务，将接着上回异常终止处继续抓取」。
  • 与 Scrapy 的取舍写在最后：feapder 把 redis、mysql、渲染、去重、报警都封装好了，上手代价是要读懂它自己的目录约定；另外它的文档有几处和源码对不上（默认重试次数、redis 键名拼写），本篇以实测为准。

四种爬虫怎么选

feapder 内置四种爬虫，对开发者暴露的接口完全一致：都是继承一个基类，写 start_requests 下发任务、写 parse 解析数据，区别只在于任务存哪儿、能不能续爬。

爬虫 依赖 断点续爬 适用场景
AirSpider 无（内存队列） 不支持 几十到几千条、跑一次就完的需求
Spider Redis 支持 海量数据、要能中途重启接着抓
TaskSpider Redis（种子表可选 MySQL） 支持 任务清单来自表或队列，一条任务对应一个 URL
BatchSpider Redis + MySQL 支持 周期性重抓（如每周更新一次商品价格），要留批次记录
  • 官方给的路线是 AirSpider → Spider → BatchSpider，理由是后一个在前一个之上加东西；本篇照这个顺序写，TaskSpider 单独放在任务表一节。
  • 和 Scrapy 对照着记更省力：AirSpider 相当于一个不用写 settings 的简化版爬虫，Spider 对应 Scrapy 的 Spider + scrapy-redis，Item 与 Pipeline 是同一套思路但少了 Selector 那一层抽象，BatchSpider 在 Scrapy 里没有对应物（那是 feapder 自己加的「批次」概念）。
  • 上手前要知道的三件事：它内部用线程而不是 asyncio（并发数就是线程数）；Spider 及以上必须有 Redis；只要用到 MySQL 相关的功能（任务表、批次表、MysqlPipeline、feapder create -i）就得准备一台能连的 MySQL。

本机实测环境

实测环境：Python 3.14.10 + feapder 1.9.3（PyPI 当前版），Redis 服务端 8.2.8 跑在 WSL 上（被 feapder 那个 3.5.3 的老客户端连上照样能用），浏览器渲染用无头 Chrome 156；入库这一路连的是同一台 WSL 里的 MySQL 9.7.4 与 MongoDB 9.0.4，Windows 侧走 127.0.0.1 经 wslrelay 连进去，Mongo 那边另补装了 pymongo 4.19.0。这份清单由 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

import redis

print("python :", sys.version.split()[0])
for pkg in ("feapder", "redis", "requests", "parsel", "PyMySQL", "loguru",
            "terminal-layout", "better-exceptions", "selenium", "playwright"):
    try:
        print("%-16s: %s" % (pkg, version(pkg)))
    except Exception:
        print("%-16s: 未安装" % pkg)
c = redis.Redis(host="127.0.0.1", port=6390)
print("服务端 redis_version :", c.info("server")["redis_version"])
1
2
3
4
5
6
7
8
9
10
11
12
13
python     : 3.14.10
feapder    : 1.9.3
redis      : 3.5.3
requests   : 2.34.2
parsel     : 1.12.1
PyMySQL    : 1.2.3
loguru     : 0.7.3
terminal-layout: 3.0.0
better-exceptions: 0.3.3
selenium   : 4.52.0
playwright : 1.64.1
服务端 redis_version = 8.2.8
feapder 用的客户端 redis 包版本 = 3.5.3

redis 3.5.3 不是选出来的，是 feapder 的依赖约束 redis<4.0.0,>=2.10.6 逼出来的，这一条会带来两个坑，见下一节。

安装与版本坑

官方 README 给的是三档安装方式，区别只在 extras：

1
2
3
pip install feapder            # 精简版：无浏览器渲染、无内存去重、不能入库 mongo
pip install "feapder[render]"  # 渲染版：加了 selenium / playwright / webdriver-manager
pip install "feapder[all]"     # 完整版：再加上 pymongo / redis-py-cluster / PyExecJS / bitarray

依赖清单与静默降级

先看它到底会装什么。pip install --dry-run 只算依赖不动环境，在本机那个已经装着 redis 8.1.0 的爬虫环境里跑，把关键几项打出来：

check_downgrade.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import re
import subprocess
import sys

PIP = r"D:/ProgramData/anaconda3/envs/spider/python.exe"

out = subprocess.run([PIP, "-m", "pip", "install", "--dry-run", "feapder"],
                     capture_output=True, text=True, encoding="utf-8",
                     errors="replace").stdout
line = next(l for l in out.splitlines() if l.startswith("Would install"))
print("Would install 里与爬虫相关的包：")
for tok in line.split()[2:]:
    if re.match(r"(redis|requests|parsel|urllib3|lxml|beautifulsoup4|bs4|feapder)-", tok):
        print("   ", tok)
print("该环境现有 redis:", subprocess.run(
    [PIP, "-c", "import redis; print(redis.__version__)"],
    capture_output=True, text=True).stdout.strip())
1
2
3
4
5
6
Would install 里与爬虫相关的包：
    bs4-0.0.2
    feapder-1.9.3
    redis-3.5.3
    requests-2.34.2
该环境现有 redis: 8.1.0

Would install 原本是一整行（475 个字符），这里只截出与爬虫相关的几项。那个 redis-3.5.3 就是降级：环境里本来是 8.1.0，pip 只说「Would install」，一个 warning 都不给。所以正确做法是给 feapder 单开一个环境：

1
2
python -m venv E:\Desktop\Test\feapder-env
E:\Desktop\Test\feapder-env\Scripts\python.exe -m pip install feapder

Requires-Dist 的完整原文（importlib.metadata.requires("feapder") 取出来的）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
better-exceptions>=0.2.2
DBUtils>=2.0
parsel>=1.5.2
PyMySQL>=0.9.3
redis<4.0.0,>=2.10.6
requests>=2.22.0
bs4>=0.0.1
ipython>=7.14.0
cryptography>=3.3.2
urllib3>=1.25.8
loguru>=0.5.3
influxdb>=5.3.1
pyperclip>=1.8.2
terminal-layout>=2.1.3
bitarray>=1.5.3; extra == "all"
PyExecJS>=1.5.1; extra == "all"
pymongo>=3.10.1; extra == "all"
redis-py-cluster>=2.1.0; extra == "all"
webdriver-manager>=4.0.0; extra == "all"
playwright; extra == "all"
selenium>=3.141.0; extra == "all"
webdriver-manager>=4.0.0; extra == "render"
playwright; extra == "render"
selenium>=3.141.0; extra == "render"

distutils 不见了

redis 3.5.3 里还在用 from distutils.version import StrictVersion，而 Python 3.12 起标准库删掉了 distutils。新环境里装完直接 import 会炸在这里：

1
E:\Desktop\Test\feapder-env\Scripts\python.exe -X utf8 -c "import feapder"
1
2
3
4
5
6
7
8
9
10
  File "E:\Desktop\Test\feapder-env\Lib\site-packages\feapder\utils\tools.py", line 44, in <module>
    import redis
  File "E:\Desktop\Test\feapder-env\Lib\site-packages\redis\__init__.py", line 1, in <module>
    from redis.client import Redis, StrictRedis
  File "E:\Desktop\Test\feapder-env\Lib\site-packages\redis\client.py", line 12, in <module>
    from redis.connection import (ConnectionPool, UnixDomainSocketConnection,
                                  SSLConnection)
  File "E:\Desktop\Test\feapder-env\Lib\site-packages\redis\connection.py", line 2, in <module>
    from distutils.version import StrictVersion
ModuleNotFoundError: No module named 'distutils'

报错在 redis，不在 feapder，容易被带偏去换 redis 版本。真正的修法是装上 setuptools，它会往 sys.path 里挂一个 distutils 兼容层：

1
E:\Desktop\Test\feapder-env\Scripts\python.exe -m pip install setuptools
1
2
setuptools 84.2.1
distutils -> E:\Desktop\Test\feapder-env\Lib\site-packages\setuptools\_distutils\__init__.py

装上之后再 import feapder 就通了。用 python -m venv 新建的环境从 3.12 起不再自带 setuptools，所以这一步不能省。想复现这个坑，把兼容层关掉即可：

1
2
$env:SETUPTOOLS_USE_DISTUTILS = "stdlib"
python -X utf8 -c "import feapder"
1
2
3
  File "E:\Desktop\Test\feapder-env\Lib\site-packages\redis\connection.py", line 2, in <module>
    from distutils.version import StrictVersion
ModuleNotFoundError: No module named 'distutils'

精简版缺什么

feapder[all] 里那几个 extras 不是「性能增强」，是功能开关。只装精简版时，mongo 管道连导入都过不了：

1
python -X utf8 -c "import feapder.pipelines.mongo_pipeline"
1
2
3
4
5
  File "E:\Desktop\Test\feapder-env\Lib\site-packages\feapder\pipelines\mongo_pipeline.py", line 12, in <module>
    from feapder.db.mongodb import MongoDB
  File "E:\Desktop\Test\feapder-env\Lib\site-packages\feapder\db\mongodb.py", line 14, in <module>
    import pymongo
ModuleNotFoundError: No module named 'pymongo'

浏览器渲染同理，而且它比文档写的更严格：feapder.utils.webdriver 这个包会把 Selenium 和 Playwright 两种驱动一起导入，只装 selenium 也过不去：

1
python -X utf8 -c "from feapder.utils.webdriver.selenium_driver import SeleniumDriver"
1
2
3
4
selenium 4.52.0
desired_capabilities 还在
feapder SeleniumDriver -> ModuleNotFoundError No module named 'playwright'
RENDER_DOWNLOADER 可用

两个都补齐之后，四个模块才全是 OK：

1
2
3
4
OK   feapder.utils.webdriver
OK   feapder.utils.webdriver.selenium_driver
OK   feapder.utils.webdriver.playwright_driver
OK   feapder.network.downloader

结论：要用渲染就照 pip install "feapder[render]" 装；只装 selenium 是不够的，因为 feapder 的 webdriver 包会一起导入 Playwright。

怎么确认自己装的版本

feapder 模块没有 __version__ 属性，直接取会 AttributeError。版本号在包目录的 VERSION 文件里，或者用标准库读元数据：

feapder_version.py
1
2
3
4
5
6
7
8
from importlib.metadata import version

import feapder

print("feapder", version("feapder"))
print("有没有 __version__ :", hasattr(feapder, "__version__"))
names = [x for x in dir(feapder) if not x.startswith("_")]
print("顶层暴露的名字（%d 个）:" % len(names), names)
1
2
3
feapder 1.9.3
有没有 __version__ : False
顶层暴露的名字（23 个）: ['AirSpider', 'ArgumentParser', 'BaseParser', 'BatchParser', 'BatchSpider', 'Item', 'Request', 'Response', 'Spider', 'TaskParser', 'TaskSpider', 'UpdateItem', 'buffer', 'core', 'db', 'dedup', 'network', 'os', 'pipelines', 're', 'setting', 'sys', 'utils']

配置键与默认值

feapder 的全部配置都在 feapder.setting 里，共 103 个大写键。挑最常用的打一遍：

setting_defaults.py
1
2
3
4
5
6
7
8
9
10
11
12
from feapder import setting

keys = [k for k in dir(setting) if k.isupper()]
print("大写配置键数量:", len(keys))
for k in ("SPIDER_THREAD_COUNT", "SPIDER_SLEEP_TIME", "SPIDER_MAX_RETRY_TIMES",
          "REQUEST_TIMEOUT", "ITEM_MAX_CACHED_COUNT", "ITEM_UPLOAD_BATCH_MAX_SIZE",
          "ITEM_UPLOAD_INTERVAL", "KEEP_ALIVE", "LOG_LEVEL", "LOG_COLOR",
          "ITEM_PIPELINES", "REQUEST_FILTER_ENABLE", "ITEM_FILTER_ENABLE",
          "RANDOM_HEADERS", "MAKE_ABSOLUTE_LINKS", "SAVE_FAILED_REQUEST",
          "REQUEST_LOST_TIMEOUT", "COLLECTOR_TASK_COUNT", "WARNING_INTERVAL",
          "WARNING_FAILED_COUNT", "DOWNLOADER", "RENDER_DOWNLOADER", "USE_SESSION"):
    print("   %-26s = %r" % (k, getattr(setting, k, "没有这个键")))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
大写配置键数量: 103
   SPIDER_THREAD_COUNT          = 1
   SPIDER_SLEEP_TIME            = 0
   SPIDER_MAX_RETRY_TIMES       = 10
   REQUEST_TIMEOUT              = 22
   ITEM_MAX_CACHED_COUNT        = 5000
   ITEM_UPLOAD_BATCH_MAX_SIZE   = 1000
   ITEM_UPLOAD_INTERVAL         = 1
   KEEP_ALIVE                   = False
   LOG_LEVEL                    = 'DEBUG'
   LOG_COLOR                    = True
   ITEM_PIPELINES               = ['feapder.pipelines.mysql_pipeline.MysqlPipeline']
   REQUEST_FILTER_ENABLE        = False
   ITEM_FILTER_ENABLE           = False
   RANDOM_HEADERS               = True
   MAKE_ABSOLUTE_LINKS          = True
   SAVE_FAILED_REQUEST          = True
   REQUEST_LOST_TIMEOUT         = 600
   COLLECTOR_TASK_COUNT         = 32
   WARNING_INTERVAL             = 3600
   WARNING_FAILED_COUNT         = 1000
   DOWNLOADER                   = 'feapder.network.downloader.RequestsDownloader'
   RENDER_DOWNLOADER            = 'feapder.network.downloader.SeleniumDownloader'
   USE_SESSION                  = False

文档里有两处和这份默认值对不上，记的时候以源码为准：

配置 文档写法 本机 1.9.3 实测
SPIDER_MAX_RETRY_TIMES 100 10
COLLECTOR_TASK_COUNT 10 32
redis 任务表键名 {redis_key}:z_requsets（requsets 拼错） {redis_key}:z_requests
失败任务键名 {redis_key}:z_failed_requsets {redis_key}:z_failed_requests

第一个爬虫

最短的完整爬虫长这样：继承 feapder.AirSpider，写两个函数——start_requests 负责下发任务，parse 负责解析。

first_air_spider.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import feapder


class FirstSpider(feapder.AirSpider):
    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/")

    def parse(self, request, response):
        print("type(request) =", type(request).__name__)
        print("type(response) =", type(response).__name__)
        print("status_code =", response.status_code, "| url =", response.url)
        print("卡片数 =", len(response.css("div.el-card")))
        print("第一个片名 =", response.xpath("//h2/text()").extract_first())


if __name__ == "__main__":
    FirstSpider().start()

直接 python first_air_spider.py 跑，默认配置是 LOG_LEVEL="DEBUG" + LOG_COLOR=True，所以第一眼看到的都是「日志夹着 print」这副样子（ANSI 颜色已剥掉；时间戳、随机 UA 和末尾那行等待日志每次都会变）：

1
2
3
4
5
6
7
8
9
10
11
12
type(request) = Request
type(response) = Response
status_code = 200 | url = https://ssr1.scrape.center/
卡片数 = 10
第一个片名 = 霸王别姬 - Farewell My Concubine
2026-11-07 21:19:36.879 | DEBUG    | feapder.network.request:get_response:line:364 | 
                -------------- FirstSpider.parse request for ----------------
                url  = https://ssr1.scrape.center/
                method = GET
                args = {'timeout': 22, 'stream': True, 'verify': False, 'headers': {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.41 Safari/537.36'}}
2026-11-07 21:19:38.243 | DEBUG    | feapder.core.parser_control:run:line:482 | 等待任务...
2026-11-07 21:19:41.881 | INFO     | feapder.core.spiders.air_spider:run:line:114 | 无任务，爬虫结束

DEBUG 会把每一次请求的 args 完整打一遍，args 里有三样东西值得注意：timeout 取的是配置里的 REQUEST_TIMEOUT=22；verify=False（feapder 默认不校验 SSL 证书）；headers 里的 User-Agent 每次都不同（RANDOM_HEADERS=True + USER_AGENT_TYPE="chrome" 的随机效果）。

把日志调干净

开发时想要详细日志，部署时要收着。两个键控制：LOG_LEVEL 管等级，LOG_COLOR 管颜色——顺带管格式，因为它决定走哪个 handler。

air_movie.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
import time

import feapder

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


class MovieSpider(feapder.AirSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO",   # 关掉 DEBUG 的请求横幅
        LOG_COLOR=False,    # 关掉颜色，同时换成 setting.LOG_FORMAT 那种格式
        SPIDER_MAX_RETRY_TIMES=2,
    )

    def start_requests(self):
        yield feapder.Request(BASE)

    def parse(self, request, response):
        cards = response.css("div.el-card")
        print("列表页卡片数:", len(cards))
        for index, card in enumerate(cards):
            name = card.css("h2::text").extract_first()
            href = card.css("a::attr(href)").extract_first()
            score = card.css("p.score::text").extract_first().strip()
            print("列表 %02d | %-45s | %s | %s" % (index + 1, name, score, href))
            if index < 2:  # 只把前两条下发为详情任务，控制输出量
                yield feapder.Request(href, callback=self.parse_detail, name=name)

    def parse_detail(self, request, response):
        title = response.css("h2::text").extract_first()
        score = response.xpath('//p[contains(@class,"score")]//text()').extract_first().strip()
        published = response.re_first(r"(\d{4}-\d{2}-\d{2}) 上映")
        categories = response.css("button.category span::text").extract()
        print("详情 |", request.name, "->", title, "|", score, "|", published,
              "|", categories, "| 字节", len(response.text))


if __name__ == "__main__":
    t0 = time.time()
    MovieSpider(thread_count=1).start()
    print("总耗时 %.2fs" % (time.time() - t0))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
列表页卡片数: 10
列表 01 | 霸王别姬 - Farewell My Concubine                  | 9.5 | https://ssr1.scrape.center/detail/1
列表 02 | 这个杀手不太冷 - Léon                                | 9.5 | https://ssr1.scrape.center/detail/2
列表 03 | 肖申克的救赎 - The Shawshank Redemption             | 9.5 | https://ssr1.scrape.center/detail/3
列表 04 | 泰坦尼克号 - Titanic                               | 9.5 | https://ssr1.scrape.center/detail/4
列表 05 | 罗马假日 - Roman Holiday                          | 9.5 | https://ssr1.scrape.center/detail/5
列表 06 | 唐伯虎点秋香 - Flirting Scholar                     | 9.5 | https://ssr1.scrape.center/detail/6
列表 07 | 乱世佳人 - Gone with the Wind                     | 9.5 | https://ssr1.scrape.center/detail/7
列表 08 | 喜剧之王 - The King of Comedy                     | 9.5 | https://ssr1.scrape.center/detail/8
列表 09 | 楚门的世界 - The Truman Show                       | 9.0 | https://ssr1.scrape.center/detail/9
列表 10 | 狮子王 - The Lion King                           | 9.0 | https://ssr1.scrape.center/detail/10
详情 | 霸王别姬 - Farewell My Concubine -> 霸王别姬 - Farewell My Concubine | 9.5 | 1993-07-26 | ['剧情', '爱情'] | 字节 397112
详情 | 这个杀手不太冷 - Léon -> 这个杀手不太冷 - Léon | 9.5 | 1994-09-14 | ['剧情', '动作', '犯罪'] | 字节 193142
MovieSpider|2026-11-07 20:05:49,453|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
总耗时 6.00s

关掉颜色后，日志格式变成 名字|时间|文件|函数|line:行号|等级| 消息，这是 setting.LOG_FORMAT 那种；开着颜色时走 loguru，格式是 时间 | 等级 | 模块:函数:line:行号 | 消息。同一句话换个 handler 就换了长相，别以为是两个东西。

  • 详情页那两条的 name 是从列表页带过来的：feapder.Request(href, callback=self.parse_detail, name=name) 里多写一个 name=...，在 parse_detail 里用 request.name 就能取到，框架参数以外的关键字参数都是「携带数据」，不用塞进 meta。
  • href 在网页源码里是 /detail/1，取出来已经是 https://ssr1.scrape.center/detail/1，这是 MAKE_ABSOLUTE_LINKS=True 的自动补全，见「Response 解析」。
  • p.score::text 取到的是 '\n 9.5' 这种带换行的串，.strip() 是必须的（练习站的分数就写在标签末尾换行处）。
  • 结尾那句 无任务，爬虫结束 是 AirSpider 判定内存队列空了才退出的，thread_count=1 传给构造函数即可改并发；同一份代码传 thread_count=5 时详情页会并发抓。

解析函数不止一个

callback 指定回调，默认是 parse。跨类回调要连 parser_name 一起写，值为类名字符串：

1
2
3
yield feapder.Request(url, callback=self.parse_detail)          # 同类回调
yield feapder.Request(url, callback="parse_detail")             # 回调名写成字符串也行
yield feapder.Request(url, parser_name="OtherSpider", callback="parse_detail")  # 跨类
  • 回调链上还能下发新任务，写法与 start_requests 一致，yield feapder.Request(...) 即可。
  • start_callback / end_callback / exception_request / failed_request / validate / download_midware 这几个钩子在 BaseParser 里，任何爬虫都能重写，见「中间件与校验」。
  • 想中途收工就调 self.stop_spider()，可以在 parse 里按条件调用。

命令行脚手架

pip install feapder 会带上一个 feapder 命令，不带参数就打印帮助：

1
feapder
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
███████╗███████╗ █████╗ ██████╗ ██████╗ ███████╗██████╗
██╔════╝██╔════╝██╔══██╗██╔══██╗██╔══██╗██╔════╝██╔══██╗
█████╗  █████╗  ███████║██████╔╝██║  ██║█████╗  ██████╔╝
██╔══╝  ██╔══╝  ██╔══██║██╔═══╝ ██║  ██║██╔══╝  ██╔══██╗
██║     ███████╗██║  ██║██║     ██████╔╝███████╗██║  ██║
╚═╝     ╚══════╝╚═╝  ╚═╝╚═╝     ╚═════╝ ╚══════╝╚═╝  ╚═╝

Version: 1.9.3
Document: https://feapder.com

Usage:
  feapder <command> [options] [args]

Available commands:
  create        create project、spider、item and so on
  retry         retry failed request or item
  shell         debug response
  zip           zip project

create 的子命令

1
feapder create -h
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
usage: python.exe -m feapder.commands.cmdline [-h] [-p ] [-s ] [-i ] [-t ]
                                              [-init] [-j] [-sj] [-c]
                                              [--params] [--setting] [--host ]
                                              [--port ] [--username ]
                                              [--password ] [--db ]

生成器

options:
  -h, --help        show this help message and exit
  -p, --project     创建项目 如 feapder create -p <project_name>
  -s, --spider      创建爬虫 如 feapder create -s <spider_name>
  -i, --item        创建item 如 feapder create -i <table_name> 支持模糊匹配 如 feapder
                    create -i %table_name%
  -t, --table       根据json创建表 如 feapder create -t <table_name>
  -init             创建__init__.py 如 feapder create -init
  -j, --json        创建json
  -sj, --sort_json  创建有序json
  -c, --cookies     创建cookie
  --params          解析地址中的参数
  --setting         创建全局配置文件feapder create --setting
  --host            mysql 连接地址
  --port            mysql 端口
  --username        mysql 用户名
  --password        mysql 密码
  --db              mysql 数据库名

-s 和 -i 会弹一个方向键菜单（AirSpider / Spider / TaskSpider / BatchSpider），菜单由 terminal-layout 实现，它要求 stdin 是真终端。在 CI、管道或重定向里跑就会直接炸：

1
feapder create -s first_spider < /dev/null
1
2
3
4
5
6
7
8
9
10
Traceback (most recent call last):
  File "...\feapder\commands\create_builder.py", line 95, in main
    choice = c.get_choice()
  File "...\terminal_layout\extensions\choice\choice.py", line 76, in get_choice
    self.ctl = LayoutCtl.quick(TableLayout, views)
  File "...\terminal_layout\ctl.py", line 210, in __init__
    self.check()
  File "...\terminal_layout\ctl.py", line 206, in check
    raise RuntimeError('terminal_layout can only run on Terminal')
RuntimeError: terminal_layout can only run on Terminal

所以自动化生成代码时不要走 feapder create，直接调用它背后的类，参数是模板类型：

gen_spider.py
1
2
3
4
5
from feapder.commands.create.create_spider import CreateSpider

for spider_type in ("AirSpider", "Spider", "TaskSpider", "BatchSpider"):
    name = "first_%s_spider" % spider_type[0].lower()
    CreateSpider().create(name, spider_type)
1
2
3
4
5
6
7
8
9
10
11
12
FirstASpider first_a_spider.py

FirstASpider 生成成功
FirstSSpider first_s_spider.py

FirstSSpider 生成成功
FirstTSpider first_t_spider.py

FirstTSpider 生成成功
FirstBSpider first_b_spider.py

FirstBSpider 生成成功

四种模板的区别只在基类、__custom_setting__ 里有没有 redis/mysql 配置、以及 __main__ 怎么启动。AirSpider 模板生成的文件全文：

first_a_spider.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
# -*- coding: utf-8 -*-
"""
Created on 2026-11-07 20:04:12
---------
@summary:
---------
@author: Loong
"""

import feapder


class FirstASpider(feapder.AirSpider):
    def start_requests(self):
        yield feapder.Request("https://spidertools.cn")

    def parse(self, request, response):
        # 提取网站title
        print(response.xpath("//title/text()").extract_first())
        # 提取网站描述
        print(response.xpath("//meta[@name='description']/@content").extract_first())
        print("网站地址: ", response.url)


if __name__ == "__main__":
    FirstASpider().start()

Spider 模板多了 redis 配置和 redis_key 参数：

first_s_spider.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import feapder


class FirstSSpider(feapder.Spider):
    # 自定义数据库，若项目中有setting.py文件，此自定义可删除
    __custom_setting__ = dict(
        REDISDB_IP_PORTS="localhost:6379", REDISDB_USER_PASS="", REDISDB_DB=0
    )

    def start_requests(self):
        yield feapder.Request("https://spidertools.cn")

    def parse(self, request, response):
        print(response)


if __name__ == "__main__":
    FirstSSpider(redis_key="xxx:xxx").start()

这份模板直接跑会连 localhost:6379，redis 不在本机或缺口令时会 AuthenticationError，要先把连接信息改成真实的。

TaskSpider 和 BatchSpider 模板还多引了 ArgumentParser，并把「下发任务」和「采集」拆成两条命令行：

first_t_spider.py（节选）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
parser = ArgumentParser(description="FirstTSpider爬虫")

parser.add_argument(
    "--start_master",
    action="store_true",
    help="添加任务",
    function=spider.start_monitor_task,
)
parser.add_argument(
    "--start_worker", action="store_true", help="启动爬虫", function=spider.start
)

parser.start()

# 通过命令行启动
# python first_t_spider.py --start_master  # 添加任务
# python first_t_spider.py --start_worker  # 启动爬虫

BatchSpider 模板的构造函数比 TaskSpider 多三个批次参数：

first_b_spider.py（节选）
1
2
3
4
5
6
7
8
9
spider = FirstBSpider(
    redis_key="xxx:xxxx",  # 分布式爬虫调度信息存储位置
    task_table="",  # mysql中的任务表
    task_keys=["id", "xxx"],  # 需要获取任务表里的字段名，可添加多个
    task_state="state",  # mysql中任务状态字段
    batch_record_table="xxx_batch_record",  # mysql中的批次记录表
    batch_name="xxx(周全)",  # 批次名字
    batch_interval=7,  # 批次周期 天为单位 若为小时 可写 1 / 24
)

命名规则

feapder create -s xxx 会先校验名字，再用「驼峰类名 + 下划线文件名」的规则同时生成两者；不合规直接不生成。

check_name.py
1
2
3
4
5
6
7
8
9
10
11
12
import re

from feapder.commands.create.create_spider import CreateSpider
from feapder.utils import tools

cs = CreateSpider()
for raw in ("first_spider", "BaiduSpider", "spider_test2", "1bad", "bad-name"):
    if re.search("^[a-zA-Z][a-zA-Z0-9_]*$", raw):
        underline = cs.cover_to_underline(raw)
        print("   %-14s -> 类名 %-14s 文件 %s.py" % (raw, tools.key2hump(underline), underline))
    else:
        print("   %-14s -> 命名不合规，不生成" % raw)
1
2
3
4
5
first_spider   -> 类名 FirstSpider     文件 first_spider.py
BaiduSpider    -> 类名 BaiduSpider     文件 baidu_spider.py
spider_test2   -> 类名 SpiderTest2     文件 spider_test2.py
1bad           -> 命名不合规，不生成
bad-name       -> 命名不合规，不生成

模板类型写错会得到：

1
ValueError: spider type error, only support AirSpider、 Spider、TaskSpider、BatchSpider

项目结构

feapder create -p demo-project 把包里的模板目录整体复制出来：

1
python -X utf8 -c "from feapder.commands.create.create_project import CreateProject; CreateProject().create('demo-project')"
1
2
3
4
5
6
7
8
9
10
11
12
demo-project 项目生成成功
生成的文件：
    CHECK_DATA.md
    README.md
    main.py
    setting.py
    __pycache__/main.cpython-314.pyc
    __pycache__/setting.cpython-314.pyc
    items/__init__.py
    items/__pycache__/__init__.cpython-314.pyc
    spiders/__init__.py
    spiders/__pycache__/__init__.cpython-314.pyc
  • main.py 是所有爬虫的统一入口，spiders/ 放爬虫、items/ 放 Item，setting.py 是全局配置（模板里整个文件都是注释掉的示例，按需取消注释）。
  • 模板目录里带了 __pycache__，拷项目时一起拷过来了；项目自带两份 markdown，其中 CHECK_DATA.md 是一份「数据审核清单」——准确性、全量性、稳定性（是否加代理、是否支持断点续跑、是否已开启报警）逐条打勾，这一份比很多教程讲的「上线前检查」更具体，值得照着过一遍。
  • 生成 spiders/ 下的爬虫文件时会自动补 __init__.py，main.py 里 from spiders import * 才import得到。

setting.py 与 main.py 入口

1
python -X utf8 -c "from feapder.commands.create.create_setting import CreateSetting; CreateSetting().create()"
1
2
3
配置文件生成成功
setting.py 是否存在: True 行数: 192
末行: # print("当前工作路径为 " + os.getcwd())

main.py 用的是 feapder 自己的 ArgumentParser，add_argument 多了一个 function= 参数，命令行命中哪个开关就调哪个函数：

main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from feapder import ArgumentParser


def crawl_movie(n):
    print("爬虫开始，参数 =", n)


def crawl_other():
    print("另一个爬虫")


parser = ArgumentParser(description="movie 爬虫")
parser.add_argument("--crawl_movie", type=int, nargs=1, choices=[1, 2],
                    help="抓电影", function=crawl_movie)
parser.add_argument("--crawl_other", action="store_true", help="抓别的",
                    function=crawl_other)
parser.start()
print("parser.start() 返回了")
1
parser.start() 返回了

不带参数时什么都不做（没有开关命中，parser.start() 直接返回）。四种调用放在一起看更直观：

1
2
3
4
python main.py --help
python main.py --crawl_movie 1
python main.py --crawl_other
python main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
usage: main.py [-h] [--crawl_movie {1,2}] [--crawl_other]

movie 爬虫

options:
  -h, --help           show this help message and exit
  --crawl_movie {1,2}  抓电影
  --crawl_other        抓别的
爬虫开始，参数 = 1
parser.start() 返回了
另一个爬虫
parser.start() 返回了
parser.start() 返回了

带 nargs=1, choices=[1, 2] 的开关是批次爬虫用的：1 下发任务、2 采集、3 重置任务，函数签名里接的就是这个整数。

需要 MySQL 的两个命令

feapder create -i 表名 是「照着 MySQL 表结构生成 Item 类」。它和 create -s 一样要真实终端（方向键菜单，非终端环境直接 RuntimeError: terminal_layout can only run on Terminal），但背后的 CreateItem 可以直接调；连上库之后，读列、生成文件一气呵成：

create_item.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
# -*- coding: utf-8 -*-
"""feapder create -i 的真身：CreateItem 照着 MySQL 表结构生成 Item 类。

命令行版要真实终端（terminal_layout），这里直接调它背后的方法。
"""

import os

import feapder.setting as setting
from feapder.commands.create.create_item import CreateItem

PW = os.environ["MYSQL_PW"]   # 口令走环境变量，不要把明文写进代码
setting.MYSQL_IP = "127.0.0.1"
setting.MYSQL_PORT = 3306
setting.MYSQL_DB = "feapder_probe"
setting.MYSQL_USER_NAME = "root"
setting.MYSQL_USER_PASS = PW

ci = CreateItem()
print("读到的列（字段名/类型/可空/默认/额外/键/注释）：")
for col in ci.select_columns("movie"):
    print("  ", col)

ci.create("movie", item_type="Item", support_dict=False)
name = "movie_item.py"
print("\n生成的文件：%s" % name)
with open(name, encoding="utf-8") as f:
    print(f.read())
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
2026-11-07 23:28:27.001 | DEBUG    | feapder.db.mysqldb:__init__:line:92 | 连接到mysql数据库 127.0.0.1 : feapder_probe
读到的列（字段名/类型/可空/默认/额外/键/注释）：
   ('crawl_time', 'datetime', 'YES', None, '', '', '')
   ('id', 'int unsigned', 'NO', None, 'auto_increment', 'PRI', '')
   ('score', 'decimal(3,1)', 'YES', None, '', '', '')
   ('title', 'varchar(255)', 'NO', None, '', 'UNI', '')

movie_item.py 生成成功

生成的文件：movie_item.py
# -*- coding: utf-8 -*-
"""
Created on 2026-11-07 23:28:27
---------
@summary:
---------
@author: Loong
"""

from feapder import Item


class MovieItem(Item):
    """
    This class was generated by feapder
    command: feapder create -i movie
    """

    __table_name__ = "movie"

    def __init__(self, *args, **kwargs):
        self.crawl_time = None
        # self.id = None
        self.score = None
        self.title = None
  • 自增主键被注释掉（# self.id = None），其余列一律初始化成 None；表名转驼峰当类名（movie → MovieItem），__table_name__ 写死成原表名。
  • 文件头两行是现场填的：Created on 取当前时间，@author 取 getpass.getuser()，可以用环境变量 FEAPDER_USER 覆盖。
  • 支持模糊匹配（feapder create -i movie* 走 information_schema.tables ... like），一次能生成一批。
  • 连不上库时它不会给你半份文件：MysqlDB.__init__ 把异常吞掉只打日志，随后 connect_pool 属性不存在，报的是 AttributeError（原文见「Item 与数据管道」的 MysqlPipeline 一节）。

feapder zip 少了目录参数则直接打印用法：

zip_usage.py
1
2
3
4
5
6
7
8
9
# -*- coding: utf-8 -*-
"""feapder zip 少了目录参数时的行为。"""

import subprocess
import sys

r = subprocess.run([sys.executable, "-X", "utf8", "-m", "feapder.commands.cmdline", "zip"],
                   capture_output=True, text=True, encoding="utf-8", timeout=60)
print((r.stdout + r.stderr).strip()[:200])
1
2
usage: feapder zip dir_path [zip_name]
python.exe -m feapder.commands.cmdline: error: the following arguments are required: dir_path

Request 参数

feapder.Request 是对 requests.Request 的封装：requests 的参数它全收，另外加了十来个框架参数。先把默认值打出来，这一张表后面反复要用。

request_defaults.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import inspect

from feapder import Request

sig = inspect.signature(Request.__init__)
print("框架参数与默认值：")
row = []
for name, p in list(sig.parameters.items())[1:]:
    row.append("%s=%r" % (name, p.default))
for i in range(0, len(row), 3):
    print("   " + "  ".join(row[i:i + 3]))

r = Request("https://ssr1.scrape.center/")
print("实例上 make_absolute_links =", r.make_absolute_links, "（构造传 None，回落到 setting）")
print("requests_kwargs（框架兜底）:", r.requests_kwargs)
1
2
3
4
5
6
7
8
框架参数与默认值：
   retry_times=0  priority=300  parser_name=None
   callback=None  filter_repeat=True  auto_request=True
   request_sync=False  use_session=None  random_user_agent=True
   download_midware=None  is_abandoned=False  render=False
   render_time=0  make_absolute_links=None
实例上 make_absolute_links = True （构造传 None，回落到 setting）
requests_kwargs（框架兜底）: {}

priority=300 会被原样写成 redis zset 的 score，因此「越小越优先」不是比喻，就是字面的有序集合排序：

priority_demo.py
1
2
3
4
5
6
7
from feapder import Request

for url, kw in (("/p1", dict(priority=1)), ("/default", {}),
               ("/p300", dict(priority=300)), ("/p999", dict(priority=999))):
    r = Request("https://ssr1.scrape.center" + url, callback="parse_x", page=7, **kw)
    d = r.to_dict
    print("   score=%-8r member=%s" % (r.priority, d))
1
2
3
4
score=1        member: {'url': 'https://ssr1.scrape.center/p1', 'priority': 1, 'callback': 'parse_x', 'make_absolute_links': True, 'page': 7}
score=300      member: {'url': 'https://ssr1.scrape.center/default', 'callback': 'parse_x', 'make_absolute_links': True, 'page': 7}
score=300      member: {'url': 'https://ssr1.scrape.center/p300', 'priority': 300, 'callback': 'parse_x', 'make_absolute_links': True, 'page': 7}
score=999      member: {'url': 'https://ssr1.scrape.center/p999', 'priority': 999, 'callback': 'parse_x', 'make_absolute_links': True, 'page': 7}

注意第二行：等于默认值的框架字段会被 to_dict 剔掉，所以 priority=300 和不传 priority 生成的 member 是同一个字符串，这两条任务在 zset 里会互相覆盖——「同 URL 同参数」天然只有一份。

携带参数与指纹

任意关键字参数都会挂到 request 上，取值时用同名的属性，类型不变：

carry_params.py
1
2
3
4
5
6
7
8
9
10
11
12
13
from feapder import Request

r = Request("https://ssr1.scrape.center/detail/1", page=2, item={"a": 1}, my_flag=True)
print("r.page =", r.page, "| r.item =", r.item, "| r.my_flag =", r.my_flag)
print("requests_kwargs 里有没有它们:", r.requests_kwargs)

base = Request("https://ssr1.scrape.center/api/movie", params={"limit": 5, "offset": 0})
other = Request("https://ssr1.scrape.center/api/movie", params={"limit": 5, "offset": 5})
same_page_diff_header = Request("https://ssr1.scrape.center/api/movie",
                                params={"limit": 5, "offset": 0}, headers={"X-Robot": "1"})
print("params 不同 -> 指纹不同 ?", base.fingerprint != other.fingerprint)
print("只有 headers 不同 -> 指纹不同 ?", base.fingerprint != same_page_diff_header.fingerprint)
print("指纹:", base.fingerprint)
1
2
3
4
5
r.page = 2 | r.item = {'a': 1} | r.my_flag = True
requests_kwargs 里有没有它们: {}
params 不同 -> 指纹不同 ? True
只有 headers 不同 -> 指纹不同 ? False
指纹: 1b4b6a5a2f7bb38e0a45bd0f0e2c6f7c

指纹只由 canonicalize_url(url) 加上 params/data/files/auth/cert/json 决定，headers、cookies、timeout、携带参数一概不进。这条规则有个直接影响：靠请求头区分翻页的接口会被判成同一条，要翻页就把差异写进 params。

顺带两条实测细节：

  • 双下划线开头的携带参数名（如 request.__page）不会走 Python 的名字改编，feapder 是按 __dict__[key] 存的，取值也建议用 request.__dict__["__page"]，或者干脆换个不带前导双下划线的名字。
  • request.copy() 会带上全部携带参数、指纹一致但不是同一个对象。

requests 参数透传

params/data/json/headers/cookies/auth/timeout/proxies/verify/cert/allow_redirects/stream/files 全部生效，用 httpbin 的回显接口验一遍：

httpbin_pass_through.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
import feapder
from feapder import setting

setting.LOG_LEVEL = "INFO"
setting.LOG_COLOR = False

API = "https://httpbin.org"

# 1) data 走表单，method 由「有没有 data/json」自动推断
resp = feapder.Request(API + "/post", data={"k": "v", "中文": "值"}).get_response()
print("1) form =", resp.json["form"], "| 请求头 Content-Type =",
      resp.json["headers"]["Content-Type"])

# 2) json 参数
r = feapder.Request(API + "/post", json={"a": 1, "b": "x"})
print("2) json =", r.get_response().json["json"])

# 3) params 会被 urlencode
r = feapder.Request(API + "/get", params={"page": 2, "q": "电影"})
print("3) args =", r.get_response().json["args"])

# 4) 自定义 headers 之后框架就不再随机 UA
r = feapder.Request(API + "/headers", headers={"User-Agent": "MyCrawler/9.9", "X-Token": "abc"})
print("4) 回显头 =", r.get_response().json["headers"].get("User-Agent"),
      r.get_response().json["headers"].get("X-Token"))

# 5) 超时是真的会超时
r = feapder.Request(API + "/delay/3", timeout=1)
try:
    r.get_response()
except Exception as exc:
    print("5) timeout=1 ->", type(exc).__name__, str(exc)[:70])

# 6) 重定向：默认跟随，关掉就能拿到 302
r = feapder.Request(API + "/redirect/2", allow_redirects=False)
resp = r.get_response()
print("6) allow_redirects=False ->", resp.status_code, resp.headers.get("Location"), resp.history)
r = feapder.Request(API + "/redirect/2")
resp = r.get_response()
print("   默认跟随 ->", resp.status_code, "history 条数", len(resp.history), "最终", resp.url)

# 7) auth 对/错（同一个受保护地址，只换口令）
for pw in ("passwd", "wrong"):
    resp = feapder.Request(API + "/basic-auth/user/passwd", auth=("user", pw)).get_response()
    print("7) 口令 %s -> status %s | 正文前 60: %r" % (pw, resp.status_code, resp.text[:60]))
1
2
3
4
5
6
7
8
9
1) form = {'k': 'v', '中文': '值'} | Content-Type = application/x-www-form-urlencoded
2) json = {'a': 1, 'b': 'x'}
3) args = {'page': '2', 'q': '电影'}
4) 回显头 = MyCrawler/9.9 abc
5) timeout=1 -> ReadTimeout HTTPSConnectionPool(host='httpbin.org', port=443): Read timed out. (read
6) allow_redirects=False -> 302 /relative-redirect/1 []
   默认跟随 -> 200 history 条数 2 最终 https://httpbin.org/get
7) auth(passwd) -> status 200, {'authenticated': True, 'user': 'user'}
7) auth(wrong) -> status 401, {'detail': {'status': 401, 'message': 'Wrong password given for user 'username''}}

框架会给每个请求兜三个默认：timeout=22（REQUEST_TIMEOUT）、stream=True、verify=False。最后这个要留意：feapder 默认不校验 SSL 证书，自签证书站点能直接过，但也意味着默认证书不校验；要校验显式写 verify=True，cert 要传字符串或二元组，传单元素元组会 IndexError: tuple index out of range。

随机 UA 的行为也要说清楚：RANDOM_HEADERS=True 且没在 headers 里手写 User-Agent 时，每个请求随机换一个 Chrome 头；一旦自己在 headers 里写了 UA（框架会打上 custom_ua=True），随机逻辑完全不参与。

random_ua_demo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
import feapder
from feapder import setting

setting.LOG_LEVEL = "INFO"
setting.LOG_COLOR = False

API = "https://httpbin.org/headers"
ua1 = feapder.Request(API).get_response().json["headers"]["User-Agent"]
ua2 = feapder.Request(API).get_response().json["headers"]["User-Agent"]
fixed = feapder.Request(API, random_user_agent=False).get_response().json["headers"]["User-Agent"]
print("两次随机 UA 相同 ?", ua1 == ua2)
print("random_user_agent=False 用的是 DEFAULT_USERAGENT ?", fixed == setting.DEFAULT_USERAGENT)
print("setting.RANDOM_HEADERS =", setting.RANDOM_HEADERS, "| USER_AGENT_TYPE =", setting.USER_AGENT_TYPE)
1
2
3
两次随机 UA 相同 ? False
random_user_agent=False 用的是 DEFAULT_USERAGENT ? True
setting.RANDOM_HEADERS = True | USER_AGENT_TYPE = chrome

其它框架参数

  • auto_request=False：解析函数收到的 response 直接是 None（源码里 if request.auto_request: ... else: response = None），要自己调 request.get_response()。
  • request_sync=True：当前线程立刻下载，不进 redis 排队，适合有效期很短的 URL。
  • use_session=True：改走 RequestsSessionDownloader，进程内复用一个 session；请求级参数优先于全局 USE_SESSION。
  • filter_repeat=False：关掉这条请求的去重，只有 REQUEST_FILTER_ENABLE=True 时才有意义（见「去重 Dedup」）。
  • is_abandoned=True：出异常不再重试，直接进失败表；异常串里含 Invalid URL 时框架会自动把它置为 True。
  • render / render_time：浏览器渲染，见「浏览器渲染」。
  • callback / parser_name：回调定位，见下一节。

callback 可以传函数对象，也可以传函数名字符串（跨类时配合 parser_name）；parser_name 写错不会报错，而是这条请求被静默丢弃：

parser_name_typo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import feapder


class Sp(feapder.Spider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0)

    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/page/2", parser_name="写错的类名")
        yield feapder.Request("https://ssr1.scrape.center/page/3")

    def parse(self, request, response):
        print("   主 parse |", request.url)


Sp(redis_key="fp28:pn").start()
1
2
3
4
Sp|2026-11-07 20:49:45,504|scheduler.py|<lambda>|line:114|INFO|
********** feapder begin **********
   主 parse | https://ssr1.scrape.center/page/3
Sp|2026-11-07 20:49:50,518|scheduler.py|spider_end|line:494|INFO| 《Sp》爬虫结束，采集耗时 5秒

page/2 那条连下载都没发生。找不到匹配 parser 时代码里就没有 else 分支，所以「某条 URL 死活不发请求」先查 parser_name 拼写。

Response 解析

Response 继承 requests.Response，在原有属性之上加了 selector 能力。用列表页和接口页各验一遍。

response_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
import feapder
from feapder import Response, setting

setting.LOG_LEVEL = "INFO"
setting.LOG_COLOR = False

html = feapder.Request("https://ssr1.scrape.center/page/1").get_response()
api = feapder.Request("https://spa1.scrape.center/api/movie",
                      params={"limit": 5, "offset": 0}).get_response()

print("type           :", type(html).__module__ + "." + type(html).__name__)
print("是否 requests.Response 的子类:", isinstance(html, __import__("requests").Response))
print("status_code    :", html.status_code, "| code 是编码别名，类型:", type(Response.__dict__["code"]).__name__)
print("url            :", html.url)
print("elapsed        :", html.elapsed, type(html.elapsed).__name__)
print("history        :", html.history)
print("encoding       :", html.encoding, "| is_html =", html.is_html)
print("text/content   :", len(html.text), type(html.content).__name__, len(html.content))
print("css 卡片数      :", len(html.css("div.el-card")))
print("css 片名        :", html.css("h2::text").extract()[:3])
print("xpath 片名      :", html.xpath("//h2/text()").extract()[:3])
print("评分原样        :", html.css("p.score::text").extract()[:2])
print("href 自动转绝对  :", html.css("div.el-card a::attr(href)").extract_first())
print("css/xpath 混用   :", html.css("a.name").xpath("./h2/text()").extract()[:2])
print("re             :", html.re(r"<title>(.*?)</title>"))
print("re_first 命中/未命中/default:", html.re_first(r"el-card"),
      repr(html.re_first(r"nosuchclass123")), html.re_first(r"nosuchclass123", default="没有这个类"))
print("bs4            :", type(html.bs4()).__name__, "->", html.bs4().select_one("h2").text)
print("extract()      :", type(html.extract()).__name__)
print()
print("json 是属性     :", type(api.json).__name__, "| count =", api.json["count"])
print("接口首条        :", {k: api.json["results"][0][k] for k in ("id", "name", "score", "published_at")})
print("offset=10 首条 id:", feapder.Request("https://spa1.scrape.center/api/movie",
      params={"limit": 5, "offset": 10}).get_response().json["results"][0]["id"])
print("错误参数名 page=2 的首条 id:", feapder.Request("https://spa1.scrape.center/api/movie",
      params={"limit": 5, "page": 2}).get_response().json["results"][0]["id"])
try:
    api.json()
except Exception as exc:
    print("写 api.json() ->", type(exc).__name__, exc)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
type           : feapder.network.response.Response
是否 requests.Response 的子类: True
status_code    : 200 | 别名 code(状态码) 也能取: code 另有用途
url            : https://ssr1.scrape.center/page/1
elapsed        : 0:00:00.352542 timedelta
history        : []
encoding       : utf-8 | is_html = True
text/content   : 42278 bytes 42506
css 卡片数      : 10
css 片名        : ['霸王别姬 - Farewell My Concubine', '这个杀手不太冷 - Léon', '肖申克的救赎 - The Shawshank Redemption']
xpath 片名      : ['霸王别姬 - Farewell My Concubine', '这个杀手不太冷 - Léon', '肖申克的救赎 - The Shawshank Redemption']
评分原样        : ['\n                9.5', '\n                9.5']
href 自动转绝对  : https://ssr1.scrape.center/detail/1
css/xpath 混用   : ['霸王别姬 - Farewell My Concubine', '这个杀手不太冷 - Léon']
re             : ['No.1 - 电影']
re_first 命中/未命中/default: el-card None 没有这个类
bs4            : BeautifulSoup -> 霸王别姬 - Farewell My Concubine
extract()      : list

json 是属性     : dict | count = 104
接口首条        : {'id': 1, 'name': '霸王别姬', 'score': 9.5, 'published_at': '1993-07-26'}
offset=10 首条 id: 11
错误参数名 page=2 的首条 id: 1
写 api.json() -> TypeError 'dict' object is not callable

要记住的四件事：

  • response.json 是 property，直接 .json 取 dict，写 .json() 会 TypeError；这是它和 requests 的 response 最大的书写差异。
  • response.code 不是状态码，而是「解析用的编码」的别名（code = property(fget=fset=encoding)）；状态码用 status_code。
  • elapsed 是 timedelta，history 是重定向链，response.request 拿到的是 requests 内部的 PreparedRequest，不是 feapder 的 Request——要原始 request 就用解析函数第一个参数。
  • 接口页对未知参数是静默忽略：page=2 照样 200、返回的还是第一页，翻页只能用 offset/limit（与《爬虫-Scrapy》里同一站的结论一致）。

相对链接转绝对

MAKE_ABSOLUTE_LINKS=True 是全局默认，实现是正则改写 a/href、img/src、link/href、script/src 四类属性。想按请求关掉，在 Request 上传参；如果是自己拿一段 HTML 构造 response，from_text 没有这个形参，得手动清缓存：

absolute_links.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import feapder
from feapder import setting

setting.LOG_LEVEL = "INFO"
setting.LOG_COLOR = False

html = '<div><a href="/detail/1">1</a><a href="detail/2">2</a><a href="https://other.com/x">3</a></div>'
resp = feapder.Response.from_text(text=html, url="https://feapder.com/list/top")
print("from_text 默认（转绝对）:", resp.css("a::attr(href)").extract())
print("from_text 有没有该形参  :", "make_absolute_links" in feapder.Response.from_text.__code__.co_varnames)

keep = feapder.Request("https://ssr1.scrape.center/", make_absolute_links=False).get_response()
conv = feapder.Request("https://ssr1.scrape.center/").get_response()
print("Request 关掉后取到的 href:", keep.css("div.el-card a::attr(href)").extract()[:2])
print("默认转绝对取到的 href    :", conv.css("div.el-card a::attr(href)").extract()[:2])
1
2
3
默认（转绝对）: ['https://feapder.com/detail/1', 'https://feapder.com/detail/2', 'https://other.com/x']
关掉后        : ['/detail/1', 'detail/2', 'https://other.com/x']
自己 urljoin  : ['https://feapder.com/detail/1', 'https://feapder.com/list/detail/2']

乱码怎么办

Response 的解码由两个属性控制：encoding（= code，声明用什么编码）和 encoding_errors（解不出来时怎么办：strict / replace / ignore）。优先级是「自己指定 > 响应头 charset > 页面 meta charset > 猜测」。

encoding_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
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import feapder
from feapder import Response, setting

setting.LOG_LEVEL = "INFO"
setting.LOG_COLOR = False

GBK_PAGE = "<html><head><meta charset=gbk></head><body><p>中文标题GBK</p></body></html>"
BAD_BYTES = "<p>中文尾巴</p>".encode("utf-8") + b"\xff\xe5"


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/gbk":
            body, head = GBK_PAGE.encode("gbk"), "text/html"      # 头里不说编码
        else:
            body, head = BAD_BYTES, "text/html; charset=utf-8"     # 声明 utf-8 但塞坏字节
        self.send_response(200)
        self.send_header("Content-Type", head)
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


srv = HTTPServer(("127.0.0.1", 8024), Handler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
G = "http://127.0.0.1:8024"

resp = feapder.Request(G + "/gbk").get_response()
print("框架猜出来的 encoding :", resp.encoding, "| 取到 :", resp.re_first(r"<p>(.*?)</p>"))
resp.code = "latin-1"
print("改成 code='latin-1'   :", repr(resp.re_first(r"<p>(.*?)</p>")))
resp.encoding = "gbk"
print("再用 encoding='gbk'   :", repr(resp.re_first(r"<p>(.*?)</p>")))
print("code 与 encoding 是同一个 property:", Response.code is Response.encoding)

for mode in ("strict", "replace", "ignore"):
    r = feapder.Request(G + "/bad").get_response()
    r.encoding_errors = mode
    print("errors=%-8s -> %r" % (mode, r.text))
srv.shutdown()
1
2
3
gbk 解码      : 中文标题GBK
code='utf-8'  : '<html><head><meta charset=gbk></head><bod'
ignore 模式   : '<html><head><meta charset=gbk></head><bod'

这一版实验说明了两件事：code 与 encoding 是同一个开关；encoding_errors="strict" 并不会把 UnicodeDecodeError 抛到你的业务里——text 内部兜了一层 bs4.UnicodeDammit，坏字节页最终多半还是能拿到正文，所以不能拿它当「编码校验」。官方文档给的乱码处理建议（response.encoding_errors = 'ignore' 或显式指定编码）依然适用，只是报错方式不是抛异常，而是拿到替换符或丢字符。

下载缓存：调试时用 redis 少打几次站

Request.get_response(save_cached=True) 会把整个 response 存进 redis，默认 20 分钟；下一次用 get_response_from_cached() 就不联网。

response_cache.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
import time

import redis
from feapder import Request, setting

setting.LOG_LEVEL = "INFO"
setting.LOG_COLOR = False
setting.REDISDB_IP_PORTS = "127.0.0.1:6390"
setting.REDISDB_USER_PASS = ""
setting.REDISDB_DB = 0

CLIENT = redis.Redis(host="127.0.0.1", port=6390)
URL = "https://ssr1.scrape.center/page/3"

t0 = time.time()
resp = Request(URL).get_response(save_cached=True)
print("真实下载耗时 %.3fs 字节 %d" % (time.time() - t0, len(resp.text)))
keys = sorted(k.decode() for k in CLIENT.keys("response_cached*"))
for k in keys:
    print("缓存键:", k, "| 类型", CLIENT.type(k).decode(), "| TTL", CLIENT.ttl(k))

t0 = time.time()
resp2 = Request(URL).get_response_from_cached()
print("从缓存取耗时 %.3fs 字节 %d" % (time.time() - t0, len(resp2.text)))

req = Request(URL)
print("request 指纹:", req.fingerprint)
print("to_dict 的键:", sorted(req.get_response().to_dict.keys()))
for k in keys:
    CLIENT.delete(k)
print("清理后剩余:", [k.decode() for k in CLIENT.keys("response_cached*")])
1
2
3
4
5
6
真实下载耗时 0.455s 字节 42280
缓存键: response_cached:test:afdb...（32 位指纹） | 类型 string | TTL 1200
从缓存取耗时 0.003s 字节 42280
request 指纹: afdb...（32 位）
to_dict 的键: ['_content', 'cookies', 'elapsed', 'encoding', 'headers', 'status_code', 'url']
清理后剩余: []
  • 键格式是 response_cached:{前缀}:{request 指纹}，手工调用时前缀是 test；跑 Spider 家族时 Scheduler 会把它换成 redis_key，并把过期时间从 1200 改为 RESPONSE_CACHED_EXPIRE_TIME（默认 3600）。
  • 全局配置 RESPONSE_CACHED_ENABLE=True / RESPONSE_CACHED_USED=True 只对 Spider/TaskSpider/BatchSpider 生效，AirSpider 不读这两个键，只能像上面这样手工调。
  • 两个键要配对：USED=True 而 ENABLE=False 时缓存永远 miss 也不会回填，日志会反复打「无response缓存 重新下载」。
  • 缓存对象缺 raw / connection / history，response.request 也会变成 None，只适合拿正文做解析。

值日钩子：中间件、校验、失败处理

BaseParser 一共给了 8 个可重写的方法，串起来是一条流水线：download_midware → 下载 → validate → parse →（异常时 exception_request / 超次时 failed_request），外加 start_callback / end_callback。

hooks_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
import urllib.request

import feapder
from feapder.network.selector import Selector


class HooksSpider(feapder.AirSpider):
    __custom_setting__ = dict(LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False)

    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/")

    def start_callback(self):
        print("1 start_callback")

    def download_midware(self, request):
        # 请求头必须是 latin-1 能编码的字符，写中文会炸（见本节末尾）
        request.headers = {"User-Agent": "HookSpider/1.0", "X-From": "midware"}
        print("2 download_midware 加了自定义头 ->", request.headers["X-From"])
        return request

    def validate(self, request, response):
        if response.status_code != 200:
            raise Exception("状态码不是 200")
        if "霸王别姬" not in response.text:
            print("3 validate 返回 False，丢弃该请求")
            return False
        print("3 validate 通过")

    def parse(self, request, response):
        print("4 parse |", request.headers.get("X-From"), "|",
              Selector(response.text).xpath("//h2/text()")[:1])
        yield feapder.Request("https://ssr1.scrape.center/page/9",
                              callback=self.parse_custom,
                              download_midware=self.custom_midware)

    def custom_midware(self, request):
        """自定义下载：不走 requests，用标准库下，返回 (request, response)"""
        with urllib.request.urlopen(request.url, timeout=20) as f:
            body = f.read().decode("utf-8")
        print("5 自定义中间件里用 urllib 下载完，字节", len(body))
        return request, feapder.Response.from_text(text=body, url=request.url)

    def parse_custom(self, request, response):
        print("6 parse_custom 收到的 response =", type(response).__name__,
              "| 片名", response.css("h2::text").extract()[:2])

    def exception_request(self, request, response):
        print("7 exception_request", request.url)

    def failed_request(self, request, response):
        print("8 failed_request", request.url)

    def end_callback(self):
        print("9 end_callback")


if __name__ == "__main__":
    HooksSpider(thread_count=1).start()
1
2
3
4
5
6
7
1 start_callback
2 download_midware 加了自定义头 -> midware
3 validate 通过
4 parse | midware | ['霸王别姬 - Farewell My Concubine']
5 自定义中间件里用 urllib 下载完，字节 42285
6 parse_custom 收到的 response = Response | 片名 ['大话西游之大圣娶亲 - 西遊記', '霸王别姬 - Farewell My Concubine']
9 end_callback
  • download_midware 返回 request 就继续走框架下载，返回 (request, response) 二元组则框架不再下载，解析函数收到的就是你构造的 response；Response.from_text 会自动转绝对链接，Selector(response.text) 是脱离 response 时自己解析的写法。
  • 中间件还支持传「函数列表」，按顺序链式处理；在某个 Request 上传 download_midware=self.custom_midware 只有这条请求走它，其余请求仍走类里的默认中间件；传 download_midware=False 两级都不走。
  • validate 里抛异常 → 重试；返回 False → 丢弃不重试；返回 True/None → 进 parse。
  • exception_request / failed_request 的返回值必须是可迭代的（可以返回新的 request 或 item），别 return None 之外的裸对象。

顺带一个中文踩坑：请求头里写中文（{"User-Agent": "钩子蜘蛛/1.0"}）时，requests 会用 latin-1 编码头字段，直接抛'latin-1' codec can't encode characters in position 0-3: ordinal not in range(256)，而且它出现在 parse error 里、框架照样重试到 10 次才丢弃——中文要放正文或 params，别放 header。

Spider 还提供了 add_parser 做多解析器集成，但有两个实测坑：必须传类不能传实例（传实例报 TypeError: 'OtherParser' object is not callable），而且一旦用了 add_parser，主爬虫类自己就不会被注册（源码里是 if not self._parsers: self._parsers.append(self)）——「继承 Spider 写 start_requests」和「集成模式」是互斥的两种用法，混用的表现是整个爬虫一行都不执行、也不报错。

add_parser_instance.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
import feapder


class OtherParser(feapder.BaseParser):
    def parse(self, request, response):
        print("别的解析器")


class Sp(feapder.Spider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0)

    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/")

    def parse(self, request, response):
        print("主 parse")


sp = Sp(redis_key="fp30:badparser")
try:
    sp.add_parser(OtherParser())          # 传的是实例
except Exception as exc:
    print("传实例 ->", type(exc).__name__ + ":", exc)

sp2 = Sp(redis_key="fp30:badparser")
try:
    sp2.add_parser("别的名字")             # 传字符串也一样
except Exception as exc:
    print("传字符串 ->", type(exc).__name__ + ":", exc)
1
2
传实例 -> TypeError: 'OtherParser' object is not callable
传字符串 -> TypeError: 'str' object is not callable

最后一个是集成时的通用陷阱：spider.start() 继承自 TailThread，在 Python 3.12+ 上 Thread.start() 会把同进程的非守护线程 join 进来，实测表现为 start() 阻塞到抓取结束（4 秒）。要把爬虫塞进 Flask/常驻进程的话，把 spider.start() 放到独立线程里跑；AirSpider 只有 stop_spider() 和 join()，没有 stop()（写 spider.stop() 会 AttributeError: 'MyAir' object has no attribute 'stop'）。

配置优先级：三种配法谁赢

feapder 的配置有三种写法，实测优先级是 __custom_setting__ > 工作目录 setting.py > 环境变量 > 包内默认。

custom_setting_wins.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import feapder
from feapder import setting


class A(feapder.AirSpider):
    __custom_setting__ = dict(SPIDER_MAX_RETRY_TIMES=99, LOG_LEVEL="INFO", LOG_COLOR=False)


class B(feapder.AirSpider):
    __custom_setting__ = dict(SPIDER_MAX_RETRY_TIMES=3)


print("实例化 A 之前:", setting.SPIDER_MAX_RETRY_TIMES)
A()
print("实例化 A 之后:", setting.SPIDER_MAX_RETRY_TIMES)
print("实例化 B 之后:", setting.SPIDER_MAX_RETRY_TIMES)
1
2
3
实例化 A 之前: 10
实例化 A 之后: 99
实例化 B 之后: 3

__custom_setting__ 是「实例化那一刻往全局 setting 上 setattr」，它名叫 custom 但并不只作用于自己：同一个进程里先实例化的爬虫会被后实例化的覆盖。多爬虫同进程跑时，把这类参数传进构造函数（thread_count=）或干脆分进程。

  • 工作目录 setting.py 的生效机制是 feapder/setting.py 末尾一行 from setting import *，它读的是 sys.path 里那个叫 setting 的模块。
  • 因此在 spiders/ 子目录里直接运行脚本时，上级目录的 setting.py 根本读不到：ModuleNotFoundError: No module named 'setting'，所有配置退回包内默认值。
  • 官方给的三种解法实测都有效：在 setting.py 同级运行（推荐统一走 main.py）、把项目根设为 Sources Root、或给进程加 PYTHONPATH（第三条是逐字验证过的）。
  • 环境变量只对包内用 os.getenv 读的那几个键有意义：REDISDB_*、MYSQL_*、MONGO_*、LOG_LEVEL、INFLUXDB_*、TASK_ID。
  • 只要 setting.py 里写了同名键，环境变量的值就会被它吃掉（实测：环境变量 LOG_LEVEL=ERROR，setting.py 写 WARNING，最终生效的是 WARNING）。

LOG_LEVEL / LOG_COLOR 还有一层：日志器是在第一次输出日志那一刻才按当时的 setting 初始化的，之后再改 setting.LOG_LEVEL 无效。想换格式，就在爬虫类上用 __custom_setting__ 配好，或直接在程序最开头改 setting 再打第一条日志。

Spider 与 Redis

Spider 是 feapder 的分布式爬虫：任务不再放内存，而是放 Redis 的有序集合，所以能断点续爬、能多进程同时跑。它比 AirSpider 多一个必填参数 redis_key（所有键的前缀），并靠 __custom_setting__ 里的 REDISDB_IP_PORTS 连库。

movie_spider.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
import time

import feapder
import redis

REDIS = "127.0.0.1:6390"
RK = "fp15:movie"
BASE = "https://ssr1.scrape.center/"
CLIENT = redis.Redis(host="127.0.0.1", port=6390)


class MovieSpider(feapder.Spider):
    __custom_setting__ = dict(
        REDISDB_IP_PORTS=REDIS,
        REDISDB_USER_PASS="",
        REDISDB_DB=0,
        LOG_LEVEL="INFO",
        LOG_COLOR=False,
        ITEM_PIPELINES=["feapder.pipelines.console_pipeline.ConsolePipeline"],
    )

    def start_requests(self):
        yield feapder.Request(BASE)

    def parse(self, request, response):
        cards = response.css("div.el-card")
        print("[%s] 列表页卡片 %d 张" % (request.url, len(cards)))
        for index, card in enumerate(cards):
            if index >= 3:
                break
            href = card.css("a::attr(href)").extract_first()
            yield feapder.Request(href, callback=self.parse_detail)

    def parse_detail(self, request, response):
        item = feapder.Item()
        item.table_name = "movie"
        item.title = response.css("h2::text").extract_first()
        item.score = response.xpath('//p[contains(@class,"score")]//text()').extract_first().strip()
        item.crawl_at = time.strftime("%Y-%m-%d %H:%M:%S")
        yield item


if __name__ == "__main__":
    MovieSpider(redis_key=RK).start()
    print("---- 跑完后的 redis ----")
    print("剩余键:", [k.decode() for k in CLIENT.keys(RK + "*")])
1
2
3
4
5
6
7
8
9
10
11
MovieSpider|2026-11-07 20:06:48,254|scheduler.py|<lambda>|line:114|INFO|
********** feapder begin **********
[https://ssr1.scrape.center/] 列表页卡片 10 张
Thread-3|2026-11-07 20:06:50,262|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 1 条数据 到 movie
Thread-3|2026-11-07 20:06:51,266|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 1 条数据 到 movie
Thread-3|2026-11-07 20:06:52,267|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 1 条数据 到 movie
MovieSpider|2026-11-07 20:06:56,272|scheduler.py|<lambda>|line:119|INFO|
********** feapder end **********
MovieSpider|2026-11-07 20:06:56,273|scheduler.py|spider_end|line:494|INFO| 《MovieSpider》爬虫结束，采集耗时 8秒
---- 跑完后的 redis ----
剩余键: []
  • ********** feapder begin/end ********** 这一对横幅是 scheduler.py 打的，中间夹着实际抓取；《MovieSpider》爬虫结束，采集耗时 8秒 里的耗时是「本轮开始到结束」，用它判断任务是否跑完比看进程退出更直观。
  • 任务全部做完后，{redis_key} 开头的键会被清空（剩余键: []），此时再跑一次等于全新抓取。半途退出时这些键会留着，那就变成下一小节的续爬。
  • Redis 连不上时不会静默重试，start() 直接把连接错误抛出来：
spider_no_redis.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import feapder


class NoRedisSpider(feapder.Spider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        REDISDB_IP_PORTS="127.0.0.1:6399", REDISDB_USER_PASS="", REDISDB_DB=0)

    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/")

    def parse(self, request, response):
        print("不该执行")


try:
    NoRedisSpider(redis_key="fp30:noredis").start()
except Exception as exc:
    print("start() 抛出 %s: %s" % (type(exc).__name__, str(exc)[:200]))
1
start() 抛出 ConnectionError: Error 10061 connecting to 127.0.0.1:6399. 由于目标计算机积极拒绝，无法连接。.

任务防丢与断点续爬

Spider 用 zset 存任务，score 有两种含义：

  • 还没被取走的任务：score 就是请求的 priority（默认 300），越小越先被取。
  • 已经被取走、正在处理的任务：Collector 取任务的瞬间把 score 原子地改成「当前时间戳 + REQUEST_LOST_TIMEOUT」（默认 600 秒），任务做完（且数据入库后）才删除。

于是「取走的瞬间」和「做完」之间，这条任务对其他节点是不可见的——这既防抢占也防丢。下面这段一次跑完三个阶段：第一轮抓到第 8 条就 os._exit(9)（模拟 kill -9）→ 打 redis 快照 → 用同样的代码再跑一轮。

kill_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
import os
import subprocess
import sys
import time

import feapder
import redis

RK = "fp15c:movie"
BASE = "https://ssr1.scrape.center/"
CLIENT = redis.Redis(host="127.0.0.1", port=6390)


class KillSpider(feapder.Spider):
    __custom_setting__ = dict(
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0,
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        SPIDER_SLEEP_TIME=1,
        ITEM_PIPELINES=["feapder.pipelines.console_pipeline.ConsolePipeline"])

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._done = 0

    def start_requests(self):
        for i in range(1, 6):
            yield feapder.Request("%spage/%d" % (BASE, i))

    def parse(self, request, response):
        for index, card in enumerate(response.css("div.el-card")):
            if index >= 3:
                break
            yield feapder.Request(card.css("a::attr(href)").extract_first(),
                                  callback=self.parse_detail)

    def parse_detail(self, request, response):
        self._done += 1
        print("第 %d 条 | %s" % (self._done, request.url))
        item = feapder.Item()
        item.table_name = "movie"
        item.url = request.url
        yield item
        if self._done == int(os.getenv("CRASH_AT", "8")):
            print("假装进程被 kill -9：直接 os._exit(9)")
            sys.stdout.flush()
            os._exit(9)


def snapshot(tag):
    print("---- redis 快照 %s ----" % tag)
    keys = sorted(k.decode() for k in CLIENT.keys(RK + "*"))
    if not keys:
        print("   没有 %s 开头的键" % RK)
        return
    now = int(time.time())
    for k in keys:
        t = CLIENT.type(k).decode()
        if t == "zset":
            print("   %-26s zset 数量=%d" % (k, CLIENT.zcard(k)))
            for m, s in CLIENT.zrange(k, 0, 2, withscores=True):
                状态 = "等待取出" if s <= now else "取出中，%d 秒后可重做" % (s - now)
                print("      score=%d %-24s %s" % (s, 状态, m.decode()[:78]))
        elif t == "hash":
            print("   %-26s hash %s" % (k, {f.decode(): v.decode()
                                       for f, v in CLIENT.hgetall(k).items()}))


def run(crash_at):
    env = dict(os.environ, CRASH_AT=str(crash_at))
    r = subprocess.run([sys.executable, "-X", "utf8", "-u", __file__, "spider"],
                       capture_output=True, text=True, encoding="utf-8",
                       errors="replace", timeout=300, env=env)
    print("\n".join(l for l in (r.stdout or "").splitlines() if l.strip() != ""))


if len(sys.argv) > 1 and sys.argv[1] == "spider":
    KillSpider(redis_key=RK).start()
    sys.exit(0)

for k in list(CLIENT.keys(RK + "*")):
    CLIENT.delete(k)
print("=== 第一轮：抓到第 8 条硬退出 ===")
run(8)
snapshot("崩溃后")
print("=== 第二轮：同样的代码再跑一次 ===")
run(9999)
snapshot("第二轮跑完")
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
=== 第一轮：抓到第 8 条硬退出 ===
KillSpider|2026-11-07 21:44:17,476|scheduler.py|<lambda>|line:114|INFO| 
********** feapder begin **********
第 1 条 | https://ssr1.scrape.center/detail/1
第 2 条 | https://ssr1.scrape.center/detail/11
第 3 条 | https://ssr1.scrape.center/detail/12
第 4 条 | https://ssr1.scrape.center/detail/13
第 5 条 | https://ssr1.scrape.center/detail/2
第 6 条 | https://ssr1.scrape.center/detail/21
第 7 条 | https://ssr1.scrape.center/detail/22
第 8 条 | https://ssr1.scrape.center/detail/23
假装进程被 kill -9：直接 os._exit(9)
---- redis 快照 崩溃后 ----
   fp15c:movie:h_spider_status hash {'spider_start_time': '1794059057', 'heartbeat_time': '1794059072'}
   fp15c:movie:z_requests     zset 数量=8
      score=1794059662 取出中，585 秒后可重做            {'url': 'https://ssr1.scrape.center/detail/23', 'parser_name': 'KillSpider', '
      score=1794059662 取出中，585 秒后可重做            {'url': 'https://ssr1.scrape.center/detail/3', 'parser_name': 'KillSpider', 'c
      score=1794059676 取出中，599 秒后可重做            {'url': 'https://ssr1.scrape.center/detail/30', 'parser_name': 'KillSpider', '
=== 第二轮：同样的代码再跑一次 ===
MainThread|2026-11-07 21:44:37,923|scheduler.py|reset_task|line:588|INFO| 重置丢失任务完毕，共8条
KillSpider|2026-11-07 21:44:37,923|scheduler.py|<lambda>|line:114|INFO| 
********** feapder begin **********
KillSpider|2026-11-07 21:44:37,930|scheduler.py|__add_task|line:201|INFO| 检查到有待做任务 7 条，不重下发新任务，将接着上回异常终止处继续抓取
第 1 条 | https://ssr1.scrape.center/detail/23
第 2 条 | https://ssr1.scrape.center/detail/3
第 3 条 | https://ssr1.scrape.center/detail/30
第 4 条 | https://ssr1.scrape.center/detail/31
第 5 条 | https://ssr1.scrape.center/detail/32
第 6 条 | https://ssr1.scrape.center/detail/41
第 7 条 | https://ssr1.scrape.center/detail/42
第 8 条 | https://ssr1.scrape.center/detail/43
KillSpider|2026-11-07 21:44:52,945|scheduler.py|spider_end|line:494|INFO| 《KillSpider》爬虫结束，采集耗时 35秒
---- redis 快照 第二轮跑完 ----
   没有 fp15c:movie 开头的键

四件事值得注意：

  • 快照里剩下的 8 条全是「取出中」状态（score 约等于当前时间 + 600），而不是「等待取出」。原因是 COLLECTOR_TASK_COUNT=32：Collector 一次能从 zset 里预取 32 条到内存队列，所以进程崩的时候，剩余任务大多已经被搬到内存里了——这就是「预取越多，崩溃时越依赖下一轮的 reset」。
  • 第二轮开头那行 重置丢失任务完毕，共8条 就是补救动作：启动时 reset_task 把这 8 条 score 落在未来、但没有活着的节点在处理的任務重置回可取状态；紧接着 检查到有待做任务 7 条，不重下发新任务 说明 start_requests 不会再执行一遍（第 8 条已经在崩掉前做完并入库，所以只剩 7 条）。两轮加起来正好覆盖 15 条详情 + 5 个列表页。
  • 第二轮跑完，fp15c:movie 开头的键又被清空，这时重跑等于全新抓取。
  • h_spider_status 里的 heartbeat_time 是心跳，spider_start_time 在结束时会被 hget(..., is_pop=True) 弹出，用来算日志里那句「采集耗时」。

想强制重抓，用 delete_keys 在启动时清掉任务表，它支持正则：

1
2
spider = KillSpider(redis_key=RK, delete_keys="*z_requests")  # 只清任务表
spider = KillSpider(redis_key=RK, delete_keys=True)           # 该前缀下全清

两个进程同时跑

「分布式」在 feapder 里就是同一个 redis_key 启动多份进程。把 CRASH_AT 设成不可能出现的值，同一台机器上开两个进程：

1
2
python kill_spider.py &  # 进程 A
python kill_spider.py &  # 进程 B
1
2
3
4
5
6
7
8
9
10
11
---- redis 快照 并发 8 秒时 ----
   fp15b:movie:h_spider_status      hash  数量=2
      spider_start_time = 1794053344
      heartbeat_time = 1794053349
   fp15b:movie:z_requests           zset  数量=14
      score=1794053945（比现在晚 593 秒）{'url': 'https://ssr1.scrape.center/page/7', 'parser_name': 'SlowSpider', 'make_absolute_links': Tru
      score=1794053947（比现在晚 595 秒）{'url': 'https://ssr1.scrape.center/detail/13', 'parser_name': 'SlowSpider', 'callback': 'parse_deta
      score=1794053951（比现在晚 599 秒）{'url': 'https://ssr1.scrape.center/detail/21', 'parser_name': 'SlowSpider', 'callback': 'parse_deta'
两个进程都退出
A: 9 次管道输出, 累计 9 条
B: 12 次管道输出, 累计 12 条

两个进程各自打印 feapder begin/end，任务被原子地分走：这批共 21 条详情任务，A 拿到 9 条、B 拿到 12 条，加起来正好 21，没有重复抓取。并发数还受两个配置影响：

1
2
3
4
5
6
7
# COLLECTOR：从 redis 取任务到内存队列的那一层
COLLECTOR_TASK_COUNT = 32   # 每次取多少条
COLLECTOR_SLEEP_TIME = 1    # 每隔几秒取一次

# SPIDER：真正发请求的线程
SPIDER_THREAD_COUNT = 1     # 线程数，建议 <= COLLECTOR_TASK_COUNT
SPIDER_SLEEP_TIME = 0       # 每解析完一个 response 后休眠，可用 [2, 5] 表示随机区间
  • COLLECTOR_TASK_COUNT 一次取太多，分布式时一池任务会被先启动的节点整批拿走，其他节点空转；官方建议它 >= SPIDER_THREAD_COUNT。
  • 练习站有 10 分钟 CDN 缓存（Cache-Control: max-age=600），所以「二次运行更快」不代表代码优化了；多线程/多进程的耗时对比见《爬虫-多线程爬虫》。

keep_alive 常驻

keep_alive=True 时一轮任务做完不退出，等下一批任务下发（master 进程用 start_monitor_task() 持续往队列里补）：

1
python kill_spider.py        # 上一节那个爬虫，构造时传 keep_alive=True
1
2
3
4
5
6
7
OkSpider|2026-11-07 21:42:01,721|scheduler.py|<lambda>|line:114|INFO| 
********** feapder begin **********
抓到 10 张卡
OkSpider|2026-11-07 21:42:06,736|scheduler.py|<lambda>|line:119|INFO| 
********** feapder end **********
OkSpider|2026-11-07 21:42:06,736|scheduler.py|spider_end|line:494|INFO| 《OkSpider》爬虫结束，采集耗时 5秒
OkSpider|2026-11-07 21:42:06,736|scheduler.py|spider_end|line:499|INFO| 爬虫不自动结束， 等待下一轮任务...

常驻模式下任务表键不会被删（对比上一节「跑完清空」），{redis_key}:h_spider_status 会一直留着 heartbeat_time，这正是 batch_interval 能生效的前提。

失败重试与校验

feapder 的重试由「抛异常」驱动，不是由状态码驱动。先看默认行为：抓一个返回 500 的页面，不写 validate。

spider_500.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import feapder


class Raw500(feapder.Spider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False,
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0,
    )

    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/page/11")

    def parse(self, request, response):
        print("parse 进来了 | status_code =", response.status_code,
              "| 标题 =", response.xpath("//title/text()").extract_first(),
              "| 字节", len(response.text))


if __name__ == "__main__":
    Raw500(redis_key="fp23:raw500").start()
1
2
parse 进来了 | status_code = 500 | 标题 = Server Error (500) | 字节 145
Raw500|2026-11-07 20:40:21,516|scheduler.py|spider_end|line:494|INFO| 《Raw500》爬虫结束，采集耗时 6秒

response.status_code 一路透传到 parse，框架不拦。要拦就在 validate 里抛，返回值 False 则是「丢弃本次请求、不重试」：

1
2
3
def validate(self, request, response):
    if response.status_code != 200:
        raise Exception("response code not 200: %s" % response.status_code)

抛了之后进入重试链，SPIDER_MAX_RETRY_TIMES 控制次数（本机 1.9.3 默认是 10，不是文档写的 100）：

1
2
3
4
5
6
7
8
9
10
11
Thread-6|2026-11-07 20:38:10,552|parser_control.py|deal_request|line:253|ERROR|
                        -------------- ValidateSpider.parse error -------------
                        error          response code not 200: 500
                        response       <Response [500]>
                        deal request   <Request https://ssr1.scrape.center/page/11>

Thread-6|2026-11-07 20:38:10,552|parser_control.py|deal_request|line:348|INFO|
                                入库 等待重试
                                url     https://ssr1.scrape.center/page/11
                                重试次数 1
                                最大允许重试次数 2

网络类错误也一样，只是 error 换成了 requests 的原始异常文本，response 是 None：

1
2
3
4
5
Thread-6|2026-11-07 20:37:59,617|parser_control.py|deal_request|line:253|ERROR|
                        -------------- FailSpider.parse error -------------
                        error          HTTPSConnectionPool(host='no-such-host-for-feapder.invalid', port=443): Max retries exceeded with url: / (Caused by NameResolutionError("HTTPSConnection(host='no-such-host-for-feapder.invalid', port=443): Failed to resolve 'no-such-host-for-feapder.invalid' ([Errno 11001] getaddrinfo failed)"))
                        response       None
                        deal request   <Request https://no-such-host-for-feapder.invalid/>

超过次数就落进失败表，SAVE_FAILED_REQUEST=True（默认）时可在 redis 里查到：

1
2
3
---- redis 快照 重试跑完 ----
   fp23:retry:z_failed_requests             zset 数量=1
        score=300 {'url': 'https://no-such-host-for-feapder.invalid/', 'method': 'GET', 'retry_times': 3, 'parser_name': 'FailSpider', 'filter_repeat': False, 'make_abs

成员里带着 retry_times: 3，排查时能直接看出它重试了几次。相关配置：

1
2
3
4
SAVE_FAILED_REQUEST = True    # 失败任务写入 {redis_key}:z_failed_requests
RETRY_FAILED_REQUESTS = False # 启动时把失败表重新下发为任务（想「重跑昨天的失败」就打开）
RETRY_FAILED_ITEMS = False    # 同上，针对入库失败的 item
WARNING_FAILED_COUNT = 1000   # 失败数超过它就报警

单条任务想放弃重试，用 feapder.Request(url, is_abandoned=True)；想拿到失败任务做兜底，重写 failed_request(request, response)，它的返回值必须可迭代（可以返回新的 request 或 item）。

batch_interval 要配合常驻才生效

Spider 构造参数里有个 batch_interval（单位天），文档说「二次运行时若间隔不足会自动退出」。本机实测：普通跑完一次是拦不住的——源码里 spider_end() 先调 record_end_time()（只有 batch_interval 非 0 才写 spider_end_time），紧接着 if not keep_alive: delete_tables(h_spider_status) 又把存这个时间的 hash 删掉了，等于自己把证据擦干净。

下面这段自己起子进程把两回都跑一遍：第一回带 keep_alive=True（跑完一轮不删键，18 秒后由父进程杀掉），第二回只带 batch_interval=7。

keep_alive_batch.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
import subprocess
import sys
import time

import redis
import feapder

RK = "fp33:batch"
CLIENT = redis.Redis(host="127.0.0.1", port=6390)


class OkSpider(feapder.Spider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0)

    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/page/2")

    def parse(self, request, response):
        print("抓到 %d 张卡" % len(response.css("div.el-card")))


if len(sys.argv) > 1 and sys.argv[1] == "round1":
    OkSpider(redis_key=RK, batch_interval=7, keep_alive=True).start()
elif len(sys.argv) > 1 and sys.argv[1] == "round2":
    OkSpider(redis_key=RK, batch_interval=7).start()
else:
    for k in list(CLIENT.keys(RK + "*")):
        CLIENT.delete(k)
    print("第 1 回：batch_interval=7 + keep_alive=True，跑完一轮后由外部杀掉")
    p = subprocess.Popen([sys.executable, "-X", "utf8", "-u", __file__, "round1"],
                         stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
                         encoding="utf-8", errors="replace")
    time.sleep(18)
    p.kill()
    print((p.stdout.read() or "").strip())
    print("redis 里留下的状态：", dict(
        (f.decode(), v.decode()) for f, v in
        CLIENT.hgetall(RK + ":h_spider_status").items()))
    print("第 2 回：同样 batch_interval=7")
    r = subprocess.run([sys.executable, "-X", "utf8", "-u", __file__, "round2"],
                       capture_output=True, text=True, encoding="utf-8",
                       errors="replace", timeout=120)
    print((r.stdout + r.stderr).strip())
1
2
3
4
5
6
7
8
9
10
11
第 1 回：batch_interval=7 + keep_alive=True，跑完一轮后由外部杀掉
OkSpider|2026-11-07 22:00:19,721|scheduler.py|<lambda>|line:114|INFO| 
********** feapder begin **********
抓到 10 张卡
OkSpider|2026-11-07 22:00:24,737|scheduler.py|<lambda>|line:119|INFO| 
********** feapder end **********
OkSpider|2026-11-07 22:00:24,737|scheduler.py|spider_end|line:494|INFO| 《OkSpider》爬虫结束，采集耗时 5秒
OkSpider|2026-11-07 22:00:24,737|scheduler.py|spider_end|line:499|INFO| 爬虫不自动结束， 等待下一轮任务...
redis 里留下的状态： {'heartbeat_time': '1794060034', 'spider_end_time': '1794060024'}
第 2 回：同样 batch_interval=7
OkSpider|2026-11-07 22:00:37,730|scheduler.py|is_reach_next_spider_time|line:524|INFO| 上次运行结束时间为 2026-11-07 22:00:24 与当前时间间隔 为 13秒, 小于规定的抓取时间间隔 7天。爬虫不执行，退出～

结论：batch_interval 只在上一轮是常驻（或异常退出、键没被清）时起作用。周期性重抓更可靠的做法是交给外部调度（crontab / feaplat / Airflow），或者用 BatchSpider——它的批次记录表就是为这个场景设计的，见「TaskSpider 与 BatchSpider」。

Item 与数据管道

yield item 之后数据才进入库流程：item 先进 ItemBuffer 的内存队列，攒够一批（ITEM_UPLOAD_BATCH_MAX_SIZE=1000）或到点（ITEM_UPLOAD_INTERVAL=1 秒）再由 pipeline 落库。这一节全部在本机实测过。

Item 是什么

item_demo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from feapder import Item, UpdateItem

item = Item()
item.table_name = "spider_data"
item.title = "霸王别姬"
item.score = 9.5
print("repr            :", repr(item))
print("to_dict         :", item.to_dict)
print("table_name      :", item.table_name)
print("item_name       :", item.item_name)
print("name_underline  :", item.name_underline)
print("item['title']   :", item["title"])
item["region"] = "中国内地"
print("下标赋值后可取   :", item.region, "| to_dict:", item.to_dict)
print("fingerprint     :", item.fingerprint)
print("未设表名时 table_name:", Item(title="x", score=1).table_name,
      "| 空 Item 的指纹:", Item().fingerprint)
1
2
3
4
5
6
7
8
9
10
11
12
repr            : <SpiderDataItem: {
    "title": "霸王别姬",
    "score": 9.5
}>
to_dict         : {'title': '霸王别姬', 'score': 9.5}
table_name      : spider_data
item_name       : SpiderDataItem
name_underline  : spider_data_item
item['title']   : 霸王别姬
下标赋值后可取   : 中国内地 | to_dict: {'title': '霸王别姬', 'score': 9.5, 'region': '中国内地'}
fingerprint     : 52e57c0694c6ecfa073428ed0fd4b9ad
未设表名时 table_name: item | 空 Item 的指纹: None
  • Item 就是个带属性的对象，to_dict 会把所有非 dunder 属性打平；表名、类名、下划线名三个属性互相联动，改一个另两个跟着变。
  • table_name 没设时取的是「类名转下划线再去掉 _item 后缀」，裸 Item 得到的是 "item"——真跑起来会往 item 表写数据，所以 item 要么显式设 table_name，要么按命名规范定义子类。
  • 指纹 fingerprint 是「所有有值字段 key+value 排序后拼起来算 md5」，全空时是 None。这条规则有两个后果：
fingerprint_demo.py
1
2
3
4
5
6
7
8
9
from feapder import Item

a, b = Item(), Item()
a.title = b.title = "同一片"
a.vote = 0      # 有值但是 falsy
b.vote = 12345
print("vote=0 与 vote=12345 指纹相同 ?", a.fingerprint == b.fingerprint)
print("   a 的 to_dict:", a.to_dict)
print("   b 的 to_dict:", b.to_dict)
1
2
3
vote=0 与 vote=12345 指纹相同 ? False
   a 的 to_dict: {'title': '同一片', 'vote': 0}
   b 的 to_dict: {'title': '同一片', 'vote': 12345}

to_dict 保留 0，但算指纹时 0 不算「有值」。也就是说 vote=0 与「没有 vote 字段」这两条数据的指纹一样，如果开了 ITEM_FILTER_ENABLE 做 item 去重，会被当成同一条。数字型字段建议用 unique_key 手工指定参与指纹的键。

指纹与表名的正确用法

unique_key_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
from feapder import Item


class MovieItem(Item):
    __unique_key__ = ["title", "url"]

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.title = None
        self.url = None
        self.crawl_time = None


m1 = MovieItem()
m1.title, m1.url, m1.crawl_time = "霸王别姬", "/detail/1", "2026-11-07 20:00:00"
m2 = MovieItem()
m2.title, m2.url, m2.crawl_time = "霸王别姬", "/detail/1", "2026-11-08 09:00:00"
m3 = MovieItem()
m3.title, m3.url, m3.crawl_time = "霸王别姬", "/detail/2", "2026-11-08 09:00:00"
print("采集时间不同、指定 unique_key 后指纹相同 ?", m1.fingerprint == m2.fingerprint)
print("url 不同时指纹相同 ?", m2.fingerprint == m3.fingerprint)

bad = MovieItem(title="传了也白传", url="/detail/9")
print("用关键字参数构造 -> to_dict:", bad.to_dict, "| 指纹:", bad.fingerprint)
1
2
3
采集时间不同、指定 unique_key 后指纹相同 ? True
url 不同时指纹相同 ? False
用关键字参数构造 -> to_dict: {'title': None, 'url': None, 'crawl_time': None} | 指纹: None

最后一行是新手最容易踩的：模板生成的 Item 子类在 __init__ 里先把每个字段置 None，super().__init__(**kwargs) 收到的关键字参数会被这几行覆盖掉，所以 MovieItem(title="x") 是无效的，只能 item = MovieItem() 再逐个赋值（或者用父类 Item 直接构造，父类不覆盖）。

其它几个方法：

item_api_demo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from feapder import Item

n = Item()
n.table_name = "tb_movie"
print("设 table_name='tb_movie' -> item_name:", n.item_name, "| name_underline:", n.name_underline)
p = Item()
p.item_name = "MovieItem"
print("设 item_name='MovieItem' -> table_name:", p.table_name)

u = Item()
u.title = "A"
u.update(title="B", extra="新字段")
print("update 后        :", u.to_dict)
u.update_strict(title="C", extra="D", another="E")
print("update_strict 后 :", u.to_dict)

base = Item()
base.table_name = "spider_data"
base.title = "改过的标题"
ui = base.to_UpdateItem()
ui.update_key = "title"
print("to_UpdateItem ->", type(ui).__name__, ui.item_name, ui.update_key)
1
2
3
4
5
设 table_name='tb_movie' -> item_name: TbMovieItem | name_underline: tb_movie_item
设 item_name='MovieItem' -> table_name: movie
update 后        : {'title': 'B', 'extra': '新字段'}
update_strict 后 : {'title': 'C', 'extra': 'D'}
to_UpdateItem -> UpdateItem SpiderDataItem ('title',)

update_strict 只更新已有字段（another 被丢掉），批量补字段时用它能避免手滑写出错字段名再落库成 Unknown column。UpdateItem 配合 update_key 走「插入时发现唯一索引冲突就更新」，所以表上必须有唯一索引，见下一小节的 SQL。

四种内置 pipeline

ITEM_PIPELINES 写的是类的模块路径字符串，可以配多个，item 会依次流经：

pipeline_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
import glob
import os
import time

import feapder
from feapder import Item
from feapder import setting
from feapder.pipelines import BasePipeline

WORK = r"E:\Desktop\Test\feapder-pipeline-demo"
os.makedirs(WORK, exist_ok=True)
os.chdir(WORK)

setting.LOG_LEVEL = "INFO"
setting.LOG_COLOR = False
setting.CSV_EXPORT_PATH = os.path.join(WORK, "csv_out")
setting.ITEM_PIPELINES = [
    "feapder.pipelines.console_pipeline.ConsolePipeline",
    "feapder.pipelines.csv_pipeline.CsvPipeline",
    "__main__.MyPipeline",
]
setting.ITEM_UPLOAD_INTERVAL = 0.05
setting.ITEM_UPLOAD_BATCH_MAX_SIZE = 2


class MyPipeline(BasePipeline):
    """pipeline 是单线程的批量操作，不建议在这里写网络请求"""

    def save_items(self, table, items):
        print("   自定义 save_items: table=%s, 条数=%d, 首条=%s" % (table, len(items), items[0]))
        return True

    def update_items(self, table, items, update_keys):
        print("   自定义 update_items: table=%s, update_keys=%s" % (table, update_keys))
        return True


class PipeSpider(feapder.AirSpider):
    def start_requests(self):
        yield feapder.Request("https://ssr1.scrape.center/")

    def parse(self, request, response):
        for card in response.css("div.el-card"):
            item = Item()
            item.table_name = "movie"
            item.title = card.css("h2::text").extract_first()
            item.score = card.css("p.score::text").extract_first().strip()
            yield item


t0 = time.time()
PipeSpider(thread_count=1).start()
time.sleep(1)
print("   跑完耗时 %.1fs" % (time.time() - t0))
for f in glob.glob(os.path.join(WORK, "csv_out", "*.csv")):
    print("   ----", os.path.basename(f))
    with open(f, encoding="utf-8-sig") as fh:
        for line in fh.read().splitlines()[:4]:
            print("   ", line)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
MainThread|2026-11-07 20:09:30,056|csv_pipeline.py|_ensure_csv_dir_exists|line:67|INFO| 创建CSV保存目录: C:\Users\Loong\AppData\Local\Temp\fp16_item\csv_out
Thread-2|2026-11-07 20:09:30,360|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 2 条数据 到 movie
Thread-2|2026-11-07 20:09:30,368|csv_pipeline.py|save_items|line:208|INFO| 共导出 2 条数据 到 movie.csv (文件路径: C:\Users\Loong\AppData\Local\Temp\fp16_item\csv_out\movie.csv)
   自定义 save_items: table=movie, 条数=2, 首条={'title': '霸王别姬 - Farewell My Concubine', 'score': '9.5'}
Thread-2|2026-11-07 20:09:30,368|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 2 条数据 到 movie
Thread-2|2026-11-07 20:09:30,372|csv_pipeline.py|save_items|line:208|INFO| 共导出 2 条数据 到 movie.csv (文件路径: C:\Users\Loong\AppData\Local\Temp\fp16_item\csv_out\movie.csv)
   自定义 save_items: table=movie, 条数=2, 首条={'title': '肖申克的救赎 - The Shawshank Redemption', 'score': '9.5'}
...
PipeSpider|2026-11-07 20:09:35,059|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
   跑完耗时 6.0s
   ---- movie.csv
    title,score
    霸王别姬 - Farewell My Concubine,9.5
    这个杀手不太冷 - Léon,9.5
    肖申克的救赎 - The Shawshank Redemption,9.5
  • 三个 pipeline 依次收到同一批数据：ConsolePipeline 只是打日志（调试首选），CsvPipeline 落 UTF-8 BOM 的 csv（用 utf-8-sig 读才对，Excel 直接双击不乱码），MyPipeline 是自定义。
  • 因为把 ITEM_UPLOAD_BATCH_MAX_SIZE 调成 2，所以每批 2 条，items 参数是「一个 list 里装着多个 dict」，不是单条——批量落库才是 pipeline 的正确姿势，别在里面写逐条网络请求。
  • 自定义 pipeline 必须继承 BasePipeline，否则在初始化阶段就被拦下：
bad_pipeline.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from feapder import setting
from feapder.buffer.item_buffer import ItemBuffer


class BadPipeline(object):
    def save_items(self, table, items):
        return True


setting.ITEM_PIPELINES = ["__main__.BadPipeline"]
try:
    ItemBuffer(redis_key="fp30:badpipe")
except Exception as exc:
    print("%s: %s" % (type(exc).__name__, exc))

setting.ITEM_PIPELINES = ["feapder.pipelines.console_pipeline.ConsolePipeline"]
ib = ItemBuffer(redis_key="fp30:okpipe")
print("继承 BasePipeline 就正常:", [type(p).__name__ for p in ib._pipelines])
1
2
ValueError: __main__.BadPipeline 需继承 feapder.pipelines.BasePipeline
继承 BasePipeline 就正常: ['ConsolePipeline']
  • save_items 返回 False 表示这批没落成功，框架不会把这批数据记进去重库，并且会重试（EXPORT_DATA_MAX_RETRY_TIMES=10、EXPORT_DATA_MAX_FAILED_TIMES=10）。
  • Item 级还能覆盖全局：__pipelines__ = [CsvPipeline()] 写在自己的类里，这条 item 就只走指定的 pipeline。pre_to_db() 是入库前回调，在 ItemBuffer.put_item 里被调用：
feapder/buffer/item_buffer.py（源码节选）
1
2
3
4
5
6
def put_item(self, item):
    if isinstance(item, Item):
        # 入库前的回调
        item.pre_to_db()

    self._items_queue.put(item)

MysqlPipeline 会发什么 SQL

这一节全部连真库跑过：MySQL 9.7.4 与 MongoDB 9.0.4 都在 WSL 里，Windows 侧用 127.0.0.1 经 wslrelay 连进去；feapder 走的是 PyMySQL 1.2.3，Mongo 那边补装了 pymongo 4.19.0。沙箱库统一叫 feapder_probe。先看 SQL 是怎么拼的（这段不连库，to_sql 离线生成）。

sql_demo.py
1
2
3
4
5
6
7
8
9
from feapder import Item

item = Item()
item.table_name = "movie"
item.title = "霸王别姬"
item.score = 9.5
item.crawl_time = "2026-11-07 20:00:00"
print("普通插入 :", item.to_sql())
print("带更新   :", item.to_sql(auto_update=True, update_columns=("score", "crawl_time")))
1
2
3
普通插入 : insert into `movie` (`title`, `score`, `crawl_time`) values ('霸王别姬', 9.5, '2026-11-07 20:00:00')

带更新   : insert into `movie` (`title`, `score`, `crawl_time`) values ('霸王别姬', 9.5, '2026-11-07 20:00:00') on duplicate key update score=values(score), crawl_time=values(crawl_time)
mysql_pipeline.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 -*-
"""MysqlPipeline 真落库：建沙箱库表 -> feapder 爬虫 yield Item -> 回查行数与内容。

口令只从环境变量 MYSQL_PW 读，绝不打印。
"""

import os

import pymysql
import feapder
from feapder import setting

PW = os.environ["MYSQL_PW"]   # 口令走环境变量，不要把明文写进代码
DB = "feapder_probe"

# 1) 沙箱库与表（title 上建唯一索引，供后面 UpdateItem 用）
admin = pymysql.connect(host="127.0.0.1", port=3306, user="root", password=PW,
                        charset="utf8mb4", connect_timeout=8)
with admin.cursor() as cur:
    cur.execute("create database if not exists `%s` default charset utf8mb4" % DB)
    cur.execute("drop table if exists `%s`.`movie`" % DB)
    cur.execute("""
        create table `%s`.`movie` (
          `id` int unsigned not null auto_increment primary key,
          `title` varchar(255) not null,
          `score` decimal(3,1) default null,
          `crawl_time` datetime default null,
          unique key `uk_title` (`title`)
        ) engine=InnoDB default charset utf8mb4
    """ % DB)
admin.close()
print("沙箱表建好：", DB, ".movie（title 上有唯一索引）")

# 2) 让 feapder 连上这个库（键名是 MYSQL_IP / PORT / DB / USER_NAME / USER_PASS）
setting.MYSQL_IP = "127.0.0.1"
setting.MYSQL_PORT = 3306
setting.MYSQL_DB = DB
setting.MYSQL_USER_NAME = "root"
setting.MYSQL_USER_PASS = PW
setting.ITEM_PIPELINES = ["feapder.pipelines.mysql_pipeline.MysqlPipeline"]


class MovieSpider(feapder.AirSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False)

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

    def parse(self, request, response):
        for card in response.css("div.el-card"):
            item = feapder.Item()
            item.table_name = "movie"
            item.title = card.css("a.name").xpath("./h2/text()").extract_first()
            item.score = float((card.css("p.score::text").extract_first() or "0").strip())
            item.crawl_time = "2026-11-07 22:30:00"
            yield item


MovieSpider(thread_count=2).start()

# 3) 回查
conn = pymysql.connect(host="127.0.0.1", port=3306, user="root", password=PW,
                       db=DB, charset="utf8mb4", connect_timeout=8)
with conn.cursor() as cur:
    cur.execute("select count(*), min(score), max(score) from movie")
    print("回查 count/min/max:", cur.fetchone())
    cur.execute("select id, title, score, crawl_time from movie order by id limit 3")
    for row in cur.fetchall():
        print("  行:", row)
conn.close()
1
2
3
4
5
6
7
沙箱表建好： feapder_probe .movie（title 上有唯一索引）
Thread-2|2026-11-07 23:01:26,947|mysql_pipeline.py|save_items|line:45|INFO| 共导出 10 条数据 到 movie, 重复 0 条
MovieSpider|2026-11-07 23:01:30,906|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
回查 count/min/max: (10, Decimal('9.0'), Decimal('9.5'))
  行: (1, '霸王别姬 - Farewell My Concubine', Decimal('9.5'), datetime.datetime(2026, 11, 7, 22, 30))
  行: (2, '这个杀手不太冷 - Léon', Decimal('9.5'), datetime.datetime(2026, 11, 7, 22, 30))
  行: (3, '肖申克的救赎 - The Shawshank Redemption', Decimal('9.5'), datetime.datetime(2026, 11, 7, 22, 30))

同一张表上把「重复数据」「UpdateItem」「没有唯一索引」「replace into」四种情况一次跑完：

mysql_behaviors.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
# -*- coding: utf-8 -*-
"""MysqlPipeline 的真实行为：普通 Item 撞唯一索引、UpdateItem 更新、无唯一索引插成新行、replace into 换 id。"""

import os

import pymysql
import feapder
from feapder import setting

PW = os.environ["MYSQL_PW"]   # 口令走环境变量，不要把明文写进代码
DB = "feapder_probe"
TITLE = "霸王别姬 - Farewell My Concubine"


def conn():
    return pymysql.connect(host="127.0.0.1", port=3306, user="root", password=PW,
                           database=DB, charset="utf8mb4", connect_timeout=8,
                           autocommit=True)


def query(text, q):
    with conn() as c, c.cursor() as cur:
        cur.execute(q)
        print("%s ->" % text, cur.fetchall())


def reset_tables():
    with conn() as c, c.cursor() as cur:
        for name, extra in (("movie", "unique key `uk_title` (`title`)"),
                            ("movie_nokey", ""),
                            ("movie_rep", "unique key `uk_title` (`title`)")):
            cur.execute("drop table if exists `%s`" % name)
            tail = ", " + extra if extra else ""
            cur.execute("create table `%s` (`id` int unsigned not null auto_increment"
                        " primary key, `title` varchar(255) not null, `score` decimal(3,1)"
                        " default null, `crawl_time` datetime default null%s)"
                        " engine=InnoDB default charset utf8mb4" % (name, tail))


setting.MYSQL_IP = "127.0.0.1"
setting.MYSQL_PORT = 3306
setting.MYSQL_DB = DB
setting.MYSQL_USER_NAME = "root"
setting.MYSQL_USER_PASS = PW
setting.ITEM_PIPELINES = ["feapder.pipelines.mysql_pipeline.MysqlPipeline"]
reset_tables()


class MovieSpider(feapder.AirSpider):
    """kind 决定产出普通 Item 还是 UpdateItem；score_of 决定写回的分数。"""
    __custom_setting__ = dict(LOG_LEVEL="INFO", LOG_COLOR=False,
                              PRINT_EXCEPTION_DETAILS=False)
    kind = "plain"
    table = "movie"
    score_of = 9.5

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

    def parse(self, request, response):
        if self.kind == "update":
            item = feapder.UpdateItem()
            item.update_key = ("score", "crawl_time")
        else:
            item = feapder.Item()
        item.table_name = self.table
        item.title = TITLE
        item.score = self.score_of
        item.crawl_time = "2026-11-07 22:30:00"
        yield item


print("== 0) 普通 Item 先插一行")
MovieSpider(thread_count=1).start()
query("movie", "select count(*) from movie")

print("\n== 1) 普通 Item 再插同一条（撞 uk_title）")
MovieSpider.kind = "dup"
MovieSpider.score_of = 96.0
MovieSpider(thread_count=1).start()
query("movie 行数与分数", "select count(*), group_concat(score) from movie")

print("\n== 2) UpdateItem 改分数（update_key=('score','crawl_time')）")
MovieSpider.kind = "update"
MovieSpider.score_of = 9.4
MovieSpider(thread_count=1).start()
query("movie 行数与分数", "select count(*), group_concat(score) from movie")

print("\n== 3) 表上没有唯一索引：同样数据跑两遍")
MovieSpider.kind = "plain"
MovieSpider.table = "movie_nokey"
MovieSpider.score_of = 9.5
MovieSpider(thread_count=1).start()
MovieSpider(thread_count=1).start()
query("movie_nokey", "select count(*), group_concat(distinct score) from movie_nokey")

print("\n== 4) auto_update=True 生成 replace into：id 会变")
item = feapder.Item()
item.table_name = "movie_rep"
item.title = TITLE
item.score = 9.5
item.crawl_time = "2026-11-07 22:30:00"
print("生成的 SQL :", item.to_sql(auto_update=True))
q = "select id, title, score from movie_rep"
with conn() as c, c.cursor() as cur:
    cur.execute(item.to_sql())
    cur.execute(q)
    print("insert 后 :", cur.fetchall())
    item.score = 9.6
    cur.execute(item.to_sql(auto_update=True))
    cur.execute(q)
    print("replace 后:", cur.fetchall())
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
== 0) 普通 Item 先插一行
Thread-2|2026-11-07 23:05:50,860|mysql_pipeline.py|save_items|line:45|INFO| 共导出 1 条数据 到 movie, 重复 0 条
MovieSpider|2026-11-07 23:05:54,826|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
movie -> ((1,),)

== 1) 普通 Item 再插同一条（撞 uk_title）
MovieSpider|2026-11-07 23:05:59,889|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
movie 行数与分数 -> ((1, '9.5'),)

== 2) UpdateItem 改分数（update_key=('score','crawl_time')）
Thread-8|2026-11-07 23:06:01,028|mysql_pipeline.py|update_items|line:74|INFO| 共更新 1 条数据 到 movie 更新字段为 ('score', 'crawl_time')
MovieSpider|2026-11-07 23:06:04,957|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
movie 行数与分数 -> ((1, '9.4'),)

== 3) 表上没有唯一索引：同样数据跑两遍
Thread-11|2026-11-07 23:06:06,068|mysql_pipeline.py|save_items|line:45|INFO| 共导出 1 条数据 到 movie_nokey, 重复 0 条
MovieSpider|2026-11-07 23:06:10,010|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
Thread-14|2026-11-07 23:06:12,095|mysql_pipeline.py|save_items|line:45|INFO| 共导出 1 条数据 到 movie_nokey, 重复 0 条
MovieSpider|2026-11-07 23:06:16,016|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
movie_nokey -> ((2, '9.5'),)

== 4) auto_update=True 生成 replace into：id 会变
生成的 SQL : replace into `movie_rep` (`title`, `score`, `crawl_time`) values ('霸王别姬 - Farewell My Concubine', 9.5, '2026-11-07 22:30:00')
insert 后 : ((1, '霸王别姬 - Farewell My Concubine', Decimal('9.5')),)
replace 后: ((2, '霸王别姬 - Farewell My Concubine', Decimal('9.6')),)

on duplicate key update 依赖表上的唯一索引才能命中，这也是官方要求 UpdateItem 必须配唯一索引的原因；没有唯一索引，同一条数据会被反复 insert 成新行。没配 MySQL 就跑 MysqlPipeline/create -i 时，实际报错是这两行（ip: None 表示连环境变量都没读到）：

1
2
3
4
5
6
7
8
9
AttributeError: 'MysqlDB' object has no attribute 'connect_pool'
2026-11-07 20:36:45.709 | ERROR    | feapder.db.mysqldb:__init__:line:78 |
            连接失败：
            ip: None
            port: 3306
            db: None
            user_name: None
            user_pass: None
            exception: utf8mb4
  • 普通 Item 撞唯一索引时不报错也不更新：make_batch_sql 的默认分支写死 insert ignore，MySQL 直接跳过重复行，affected rows 是 0，框架只在 LOG_LEVEL=DEBUG 下打一行「没有插入数据，可能全部重复」，而 save_items 照样返回 True。INFO 级别下这批数据是静默消失的——「跑完没报错但库里没数」的第一现场就在这。
  • 表上没有唯一索引时 insert ignore 形同普通 insert，同样数据跑两遍就是 20 行（上面第 3 段）。所以「不重复入库」要么靠唯一索引，要么开 ITEM_FILTER_ENABLE，不能指望 pipeline 自己去重。
  • UpdateItem.update_key 是「要更新的列」，实测 共更新 1 条数据 到 movie 更新字段为 ('score', 'crawl_time')，行数不变；值没变时 affected rows 是 0，日志也就不打这行。
  • to_sql(auto_update=True) 生成 replace into，MySQL 的实现是「删了再插」，实测自增 id 从 1 跳到 2。有外键或想保住 id 的场景别用它。

MongoPipeline 落库

MongoPipeline 要 pymongo（pip install "feapder[all]" 或单装），连接键是 MONGO_IP / MONGO_PORT / MONGO_DB。MONGO_IP 建议写 127.0.0.1：写 localhost 时 Windows 先解析到 ::1，而 mongod 只绑了 127.0.0.1，每次连接要多等两秒。三种情况一次跑完：

mongo_pipeline.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 -*-
"""MongoPipeline 真落库：无唯一索引时重复插入；有唯一索引时冲突走「更新」；UpdateItem 指定更新字段。"""

import pymongo
import feapder
from feapder import setting

DB = "feapder_probe"
COLL = "movie_mongo"

setting.MONGO_IP = "127.0.0.1"          # 写 localhost 会先试 ::1，本机要多等两秒
setting.MONGO_PORT = 27017
setting.MONGO_DB = DB
setting.ITEM_PIPELINES = ["feapder.pipelines.mongo_pipeline.MongoPipeline"]

client = pymongo.MongoClient("mongodb://127.0.0.1:27017", serverSelectionTimeoutMS=8000)
coll = client[DB][COLL]


class MovieSpider(feapder.AirSpider):
    __custom_setting__ = dict(LOG_LEVEL="INFO", LOG_COLOR=False,
                              PRINT_EXCEPTION_DETAILS=False)
    score_of = 9.5

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

    def parse(self, request, response):
        for card in response.css("div.el-card"):
            item = feapder.Item()
            item.table_name = COLL
            item.title = card.css("a.name").xpath("./h2/text()").extract_first()
            item.score = self.score_of
            yield item


class UpdateSpider(MovieSpider):
    def parse(self, request, response):
        for card in response.css("div.el-card"):
            item = feapder.UpdateItem()
            item.update_key = ("score",)
            item.table_name = COLL
            item.title = card.css("a.name").xpath("./h2/text()").extract_first()
            item.score = 6.6
            yield item


print("== 1) 集合上没有唯一索引：同样数据跑两遍")
coll.drop()
print("索引：", [ix["name"] for ix in coll.list_indexes()])
MovieSpider(thread_count=2).start()
MovieSpider.score_of = 8.8
MovieSpider(thread_count=2).start()
print("条数:", coll.count_documents({}), "| 分数种类:", sorted(coll.distinct("score")))

print("\n== 2) 先建 title 唯一索引，再跑两遍普通 Item")
coll.drop()
coll.create_index("title", unique=True)
print("索引：", [ix["name"] for ix in coll.list_indexes()])
MovieSpider.score_of = 9.5
MovieSpider(thread_count=2).start()
MovieSpider.score_of = 8.8
MovieSpider(thread_count=2).start()
print("条数:", coll.count_documents({}), "| 分数种类:", sorted(coll.distinct("score")))

print("\n== 3) UpdateItem，update_key=('score',)")
UpdateSpider(thread_count=2).start()
print("条数:", coll.count_documents({}), "| 分数种类:", sorted(coll.distinct("score")))
doc = coll.find_one({"title": "霸王别姬 - Farewell My Concubine"})
print("抽查:", {k: v for k, v in doc.items() if k != "_id"})
client.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
== 1) 集合上没有唯一索引：同样数据跑两遍
索引： []
Thread-2|2026-11-07 23:09:08,469|mongo_pipeline.py|save_items|line:42|INFO| 共导出 10 条数据到 movie_mongo,  新增 10条, 重复 0 条
MovieSpider|2026-11-07 23:09:12,412|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
Thread-6|2026-11-07 23:09:13,430|mongo_pipeline.py|save_items|line:42|INFO| 共导出 10 条数据到 movie_mongo,  新增 10条, 重复 0 条
MovieSpider|2026-11-07 23:09:17,417|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
条数: 20 | 分数种类: [8.8, 9.5]

== 2) 先建 title 唯一索引，再跑两遍普通 Item
索引： ['_id_', 'title_1']
Thread-10|2026-11-07 23:09:18,508|mongo_pipeline.py|save_items|line:42|INFO| 共导出 10 条数据到 movie_mongo,  新增 10条, 重复 0 条
MovieSpider|2026-11-07 23:09:22,479|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
Thread-14|2026-11-07 23:09:23,522|mongo_pipeline.py|save_items|line:42|INFO| 共导出 10 条数据到 movie_mongo,  新增 0条, 重复 10 条
MovieSpider|2026-11-07 23:09:27,485|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
条数: 10 | 分数种类: [9.5]

== 3) UpdateItem，update_key=('score',)
Thread-18|2026-11-07 23:09:28,536|mongo_pipeline.py|update_items|line:79|INFO| 共导出 10 条数据到 movie_mongo,  新增 0 条, 更新 10 条 更新字段为 ('score',)
UpdateSpider|2026-11-07 23:09:32,491|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
条数: 10 | 分数种类: [6.6]
抽查: {'title': '霸王别姬 - Farewell My Concubine', 'score': 6.6}
  • 集合上没有唯一索引 → 每轮都算新数据，新增 10条, 重复 0 条 打两遍，20 条。
  • 建了 title 唯一索引 → 第二轮变成 新增 0条, 重复 10 条，普通 Item 的重复数据是被丢掉的，不会被更新，和 MySQL 的 insert ignore 语义一致。
  • 要更新就得用 UpdateItem 并给出 update_key，实测 新增 0 条, 更新 10 条 更新字段为 ('score',)，条数不变、分数换成新值。

CsvPipeline / ConsolePipeline 零依赖，本地验证数据形状时优先用后两个。

去重 Dedup

feapder.dedup.Dedup 是一个可以单独使用的去重库，四种过滤器（BloomFilter=1、MemoryFilter=2、ExpireFilter=3、LiteFilter=4），默认是 BloomFilter。先把 API 语义弄清楚——官方文档里那几个 assert 只能第一次跑通过。

dedup_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
from feapder import setting
from feapder.dedup import Dedup

setting.REDISDB_IP_PORTS = "127.0.0.1:6390"
setting.REDISDB_USER_PASS = ""
setting.REDISDB_DB = 0

data = {"xxx": 123, "xxxx": "xxxx"}
datas = ["xxx", "bbb"]

b = Dedup(Dedup.BloomFilter, name="fpArt_b")
print("repr:", repr(b))
print("add(data) 第一次      ->", b.add(data))
print("get(data)            ->", b.get(data))
print("add(datas) 第一次      ->", b.add(datas))
print("add(data) 重复        ->", b.add(data))
print("add(datas) 重复        ->", b.add(datas))
print("add(['dup','dup'])   ->", b.add(["dup", "dup"]))
print("get(['newX','newX']) ->", b.get(["newX", "newX"]))

left = ["xxx", "bbb", "ccc"]
ret = b.filter_exist_data(left)
print("filter_exist_data 返回:", ret, "| 原变量:", left, "| 是同一个对象?", ret is left)

e = Dedup(Dedup.ExpireFilter, expire_time=10, name="fpArt_e")
print()
print("ExpireFilter repr:", repr(e), "| add 第一次", e.add("A"), "| add 第二次", e.add("A"))
l = Dedup(Dedup.LiteFilter)
print("LiteFilter 内层类型:", type(l.dedup).__name__, "| add 第一次", l.add("A"), "| add 第二次", l.add("A"))
try:
    Dedup(Dedup.MemoryFilter)
except Exception as exc:
    print("MemoryFilter ->", type(exc).__name__, ":", str(exc).replace("\n", " / ")[:120])
1
2
3
4
5
6
7
8
9
10
11
12
13
repr: <ScalableBloomFilter: RedisBitArray: dedup:bloomfilter:fpArt_b0>
add(data) 第一次      -> 1
get(data)            -> 1
add(datas) 第一次      -> [1, 1]
add(data) 重复        -> 0
add(datas) 重复        -> [0, 0]
add(['dup','dup'])   -> [1, 0]
get(['newX','newX']) -> [0, 1]
filter_exist_data 返回: ['ccc'] | 原变量: ['ccc'] | 是同一个对象? True

ExpireFilter repr: <ExpireSet: dedup:expire_set:fpArt_e> | add 第一次 1 | add 第二次 0
LiteFilter 内层类型: LiteFilter | add 第一次 1 | add 第二次 0
MemoryFilter -> Exception : 需要安装feapder完整版 / command: pip install "feapder[all]" / 若安装出错，参考：https://feapder.com/#/question/%E5%AE%89%E8%A3%85%E9%97%AE%E9%A2%98

读法：

  • add 返回的是「本次是否新入库」，已存在返回 0；get 返回的是「是否存在」。文档里 assert dedup.add(datas) == [1, 1] 这类断言只在库为空时成立，脚本重复跑第二遍必然 AssertionError。
  • add(['dup','dup']) 是 [1, 0]：同批里的重复项只算一条入库。get(['newX','newX']) 却是 [0, 1]——get 内部先 set() 去重再回填，所以它是「这批里第一次出现的判不存在」，不能说它等于逐条查询。
  • filter_exist_data 原地修改列表同时返回同一个对象（源码 docstring 写 @return: None 是错的）。
  • MemoryFilter 依赖第三方 bitarray，缺包时抛的不是 ModuleNotFoundError，而是 feapder 自己包装过的 Exception: 需要安装feapder完整版——排查时容易找错方向。
  • LiteFilter 就是进程内的一个 Python set：不碰 redis、name/redis_url 传了也不生效、进程退出即失效。

键名与容量：默认 BloomFilter 很贵

dedup_keys.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 time

import redis
from feapder import setting
from feapder.dedup import Dedup

setting.REDISDB_IP_PORTS = "127.0.0.1:6390"
CLIENT = redis.Redis(host="127.0.0.1", port=6390)
for k in list(CLIENT.keys("dedup:*fpArt*")):
    CLIENT.delete(k)

b = Dedup(Dedup.BloomFilter, name="fpArt_big")
t0 = time.time()
for i in range(4):
    b.add("样本%d" % i)
print("只写 4 条，耗时 %.3fs" % (time.time() - t0))
inner = b.dedup
print("内层位图: num_slices=%s num_bits=%s" % (
    getattr(inner.filters[0], "num_slices", "?"),
    getattr(inner.filters[0], "num_bits", "?")))
for k in sorted(x.decode() for x in CLIENT.keys("dedup:*fpArt*")):
    print("   %-40s type=%-7s 字节=%d" % (k, CLIENT.type(k).decode(), CLIENT.memory_usage(k)))

e = Dedup(Dedup.ExpireFilter, expire_time=100, name="fpArt_perf")
t0 = time.time()
e.add(["数据%06d" % i for i in range(10000)])
one = time.time() - t0
t0 = time.time()
for i in range(10):
    e.add(["批次%02d_数据%06d" % (j, i * 10000 + j) for j in range(10000)])
ten = time.time() - t0
print("ExpireFilter 写 1 万 %.3fs，写 10 万（10 批）%.3fs" % (one, ten))
print("   键类型:", CLIENT.type("dedup:expire_set:fpArt_perf").decode(),
      "成员数:", CLIENT.zcard("dedup:expire_set:fpArt_perf"),
      "键级 TTL:", CLIENT.ttl("dedup:expire_set:fpArt_perf"))

l = Dedup(Dedup.LiteFilter)
t0 = time.time()
l.add(["L%06d" % i for i in range(10000)])
print("LiteFilter 写 1 万 %.3fs" % (time.time() - t0))
1
2
3
4
5
6
7
只写 4 条，耗时 0.041s
内层位图: num_slices=17 num_bits=2396264597
   dedup:bloomfilter:fpArt_big0            type=string  字节=335544368
   dedup:bloomfilter:fpArt_big0_count_cached type=string 字节=80
ExpireFilter 写 1 万 0.108s，写 10 万（10 批）0.942s
   键类型: zset 成员数: 110004 键级 TTL: -1
LiteFilter 写 1 万 0.013s

四个结论：

  • BloomFilter 的键名末尾粘着分片序号：name="fpArt_big" 的真键是 dedup:bloomfilter:fpArt_big0（源码 self.name + str(len(self.filters))，中间没分隔符），另有一个 _count_cached 计数缓存键。文档写的 dedup:bloomfilter:[name] 少这个后缀，排查时 KEYS 会找不到。
  • 默认 initial_capacity=1e8 + error_rate=1e-5 意味着位图长度 2,396,264,597 位。只写 4 条数据，redis 里那个 string 就被撑到 335MB——offset 分布在整个位图上，第一个高位 offset 就让 Redis 分配满长度。文档那句「一亿条数据占用内存约 285MB」的真实含义是「默认容量下只要用过一次就约 285MB/库」。生产上要显式给 initial_capacity，或者换 ExpireFilter/LiteFilter。
  • 写入速度：ExpireFilter 1 万条 0.098s、10 万条（拆 10 批）0.975s，LiteFilter 1 万条 0.010s——文档里的 0.26s / 3.5s 都别抄。一次 add 十万条会卡在 feapder 的 pipeline 上（拆成 10 批才跑得动），大批量要自己分批。
  • ExpireFilter 存的是 zset、score 为写入时的时间戳，键本身没有 TTL（ttl=-1）：「过期」是每次 add 时顺手 zremrangebyscore 清理，长期不写入的库会一直堆着。

在爬虫里开去重

配置只有两对，默认值现场读出来是这样：

filter_defaults.py
1
2
3
4
5
6
7
8
9
from feapder import setting
from feapder.dedup import Dedup

print("REQUEST_FILTER_ENABLE =", setting.REQUEST_FILTER_ENABLE)
print("REQUEST_FILTER_SETTING=", setting.REQUEST_FILTER_SETTING)
print("ITEM_FILTER_ENABLE    =", setting.ITEM_FILTER_ENABLE)
print("ITEM_FILTER_SETTING   =", setting.ITEM_FILTER_SETTING)
print("filter_type 常量: BloomFilter=%d MemoryFilter=%d ExpireFilter=%d LiteFilter=%d" % (
    Dedup.BloomFilter, Dedup.MemoryFilter, Dedup.ExpireFilter, Dedup.LiteFilter))
1
2
3
4
5
REQUEST_FILTER_ENABLE = False
REQUEST_FILTER_SETTING= {'filter_type': 3, 'expire_time': 2592000}
ITEM_FILTER_ENABLE    = False
ITEM_FILTER_SETTING   = {'filter_type': 1}
filter_type 常量: BloomFilter=1 MemoryFilter=2 ExpireFilter=3 LiteFilter=4

下面这段把「同一 URL 下发两次 + 四条 item 里有一条完全重复」摆在 AirSpider 上跑，用 LiteFilter（filter_type=4，不依赖 redis 也够说明效果）：

dedup_in_spider.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
import feapder
from feapder import Item

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


class DedupSpider(feapder.AirSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False,
        REQUEST_FILTER_ENABLE=True, REQUEST_FILTER_SETTING=dict(filter_type=4),
        ITEM_FILTER_ENABLE=True, ITEM_FILTER_SETTING=dict(filter_type=4),
        ITEM_PIPELINES=["feapder.pipelines.console_pipeline.ConsolePipeline"],
    )

    def start_requests(self):
        yield feapder.Request(BASE)
        yield feapder.Request(BASE)      # 同一个 URL 再下发一次

    def parse(self, request, response):
        print("   首页被请求了一次")
        for card in response.css("div.el-card")[:3]:
            item = Item()
            item.table_name = "dedup_demo"
            item.title = card.css("h2::text").extract_first()
            yield item
        # 与第一条完全相同的 item
        item = Item()
        item.table_name = "dedup_demo"
        item.title = response.css("h2::text").extract_first()
        yield item


DedupSpider(thread_count=1).start()
1
2
3
   首页被请求了一次
Thread-2|...|item_buffer.py|__dedup_items|line:211|INFO| 待入库数据 4 条， 重复 1 条，实际待入库数据 3 条
Thread-2|...|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 3 条数据 到 dedup_demo
  • 首页只被请求了一次（第二次下发被 request 去重挡掉），4 条 item 只导出 3 条，日志里那行 待入库数据 4 条， 重复 1 条 就是 item 去重的直接证据。
  • 官方文档写「AirSpider 不支持去重，因此配置文件中的去重配置无效」，在 1.9.3 上不成立：AirSpider 用的 AirSpiderRequestBuffer 与 Spider 共用同一套 Dedup 逻辑，上面这段跑的就是 AirSpider。
  • 关掉去重的逃生门是单条请求 feapder.Request(url, filter_repeat=False)，它优先于全局开关。
  • filter_type=2（MemoryFilter）在爬虫里是「0 请求直接退出」：去重库是在建 buffer 时构造的，缺 bitarray 时那条 Exception 会把整个爬虫带停。
  • 不传 name 时，BloomFilter 用的是全局库 dedup:bloomfilter:bloomfilter0，所有项目共用。换一个新爬虫跑同样的 URL，一条请求都发不出去——「重跑爬虫没数据」最常见的原因就是它。要么 REQUEST_FILTER_SETTING=dict(filter_type=1, name="项目名")，要么用 filter_type=3/4。
  • 同一进程内改 setting.REDISDB_IP_PORTS 不一定生效（RedisDB 有实例缓存），换 redis 地址要重启进程。

TaskSpider 与 BatchSpider

这两种爬虫把「要抓什么」从代码里搬到表里：TaskSpider 有一条任务表（种子队列），BatchSpider 在其上再加一层批次记录。它们都分 master（下发/监控）和 worker（消费）两种进程。

redis 做任务表：不连 MySQL 也能跑

task_spider.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
import feapder
from feapder import ArgumentParser

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

SET = dict(
    LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
    REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0,
    ITEM_PIPELINES=["feapder.pipelines.console_pipeline.ConsolePipeline"],
)


class TaskMovie(feapder.TaskSpider):
    __custom_setting__ = SET

    def add_task(self):
        """start_monitor_task 会自动调它，用来往种子表塞任务"""
        for i in (1, 2, 3):
            self._redisdb.zadd(self._task_table,
                               {"id": i, "url": "%spage/%d" % (BASE, i)})

    def start_requests(self, task):
        task_id, url = task
        print("取到种子 id=%s url=%s" % (task_id, url))
        yield feapder.Request(url, task_id=task_id)

    def parse(self, request, response):
        print("解析列表 %s -> %d 张卡（task_id=%s）" % (
            request.url, len(response.css("div.el-card")), request.task_id))
        item = feapder.Item()
        item.table_name = "task_page"
        item.task_id = request.task_id
        item.url = request.url
        yield item


if __name__ == "__main__":
    parser = ArgumentParser(description="TaskSpider 演示")
    parser.add_argument("--master", type=int, nargs=1, choices=[1],
                        help="下发种子", function=lambda _: TaskMovie(
                            redis_key="fp18:task_spider", task_table="fp18:seeds",
                            task_table_type="redis", use_mysql=False).start_monitor_task())
    parser.add_argument("--worker", action="store_true", help="消费种子",
                        function=lambda: TaskMovie(
                            redis_key="fp18:task_spider", task_table="fp18:seeds",
                            task_table_type="redis", use_mysql=False).start())
    parser.start()
    print("用法：python task_spider.py --master 1 下发种子；python task_spider.py --worker 采集")

上面这个文件不带参数运行只打印用法（parser.start() 没命中开关）：

1
用法：python task_spider.py --master 1 下发种子；python task_spider.py --worker 采集

真正跑起来分两条命令。master 下发完种子会常驻监控，每 5 秒检查一次，这里跑 12 秒后手动停掉：

1
python task_spider.py --master 1
1
2
3
4
5
6
7
MainThread|2026-11-07 20:12:50,303|task_spider.py|get_task|line:239|INFO| redis 中剩余任务0 数量过小 从redis种子任务表中取任务追加
取到种子 id=1 url=https://ssr1.scrape.center/page/1
取到种子 id=2 url=https://ssr1.scrape.center/page/2
取到种子 id=3 url=https://ssr1.scrape.center/page/3
MainThread|2026-11-07 20:12:50,305|task_spider.py|start_monitor_task|line:212|INFO| 添加任务到redis成功 共3条
MainThread|2026-11-07 20:12:55,307|task_spider.py|start_monitor_task|line:208|INFO| 无待做种子 redis中剩余任务 3
MainThread|2026-11-07 20:13:00,310|task_spider.py|start_monitor_task|line:208|INFO| 无待做种子 redis中剩余任务 3

worker 消费：

1
python task_spider.py --worker
1
2
3
4
5
6
TaskMovie|2026-11-07 20:14:50,436|scheduler.py|<lambda>|line:114|INFO|
********** feapder begin **********
解析列表 https://ssr1.scrape.center/page/1 -> 10 张卡（task_id=1）
解析列表 https://ssr1.scrape.center/page/2 -> 10 张卡（task_id=2）
Thread-3|2026-11-07 20:14:51,438|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 2 条数据 到 task_page
解析列表 https://ssr1.scrape.center/page/3 -> 10 张卡（task_id=3）

三个要点：

  • add_task 是框架在 start_monitor_task 里自动调的方便函数，它把种子写进 self._task_table；生产环境更常见的是自己写个脚本往 redis 里塞，或者从 MySQL 取。
  • start_requests(self, task) 多了一个 task 参数，它就是种子表里取出的一条（列表或带属性的对象），task_id 之类要靠 feapder.Request(url, task_id=...) 携带到 parse。
  • 上面 worker 用的是 redis 任务表 + use_mysql=False，因此不能写 yield self.update_task_batch(request.task_id)：那是要更新 MySQL 任务表 state 的。下面这段就是故意这么写，master 下发一条种子后用 Ctrl+C 停掉，再起 worker：
task_wrong_update.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
import sys

import feapder

BASE = "https://ssr1.scrape.center/"
SET = dict(LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
           REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0,
           ITEM_PIPELINES=["feapder.pipelines.console_pipeline.ConsolePipeline"])


class WrongTask(feapder.TaskSpider):
    __custom_setting__ = SET

    def add_task(self):
        self._redisdb.zadd(self._task_table, {"id": 1, "url": BASE})

    def start_requests(self, task):
        yield feapder.Request(task["url"], task_id=task["id"])

    def parse(self, request, response):
        item = feapder.Item()
        item.table_name = "wrong_task"
        item.count = len(response.css("div.el-card"))
        yield item
        yield self.update_task_batch(request.task_id)   # 误用：这张任务表在 redis，不在 mysql


mode = sys.argv[1] if len(sys.argv) > 1 else ""
if mode not in ("master", "worker"):
    print("用法：python task_wrong_update.py master   # 只下发种子")
    print("      python task_wrong_update.py worker   # 只消费任务")
    sys.exit(0)
spider = WrongTask(redis_key="fp33:wrong", task_table="fp33:seeds",
                   task_table_type="redis", use_mysql=False)
if mode == "master":
    spider.start_monitor_task()      # 下发完种子就一直监控
else:
    spider.start()
1
2
python task_wrong_update.py master   # 看到「添加任务到redis成功 共1条」后 Ctrl+C
python task_wrong_update.py worker   # 45 秒后 Ctrl+C
1
2
3
4
5
6
7
8
9
10
11
12
13
14
WrongTask|2026-11-07 21:48:39,487|scheduler.py|<lambda>|line:114|INFO| 
********** feapder begin **********
Thread-3|2026-11-07 21:48:40,489|console_pipeline.py|save_items|line:32|INFO| 【调试输出】共导出 1 条数据 到 wrong_task
Thread-3|2026-11-07 21:48:40,513|mysqldb.py|__init__|line:78|ERROR| 
            连接失败：
            ip: None
            port: 3306
            db: None
            user_name: None
            user_pass: None
            exception: utf8mb4
Thread-3|2026-11-07 21:48:40,513|mysqldb.py|add_batch|line:295|ERROR| 
                error:'MysqlDB' object has no attribute 'connect_pool'
                sql:  insert into `fp33:seeds` (`id`, `state`) values (%s, %s) on duplicate key update `id`=values(`id`), `state`=values(`state`)

worker 因为「入库不成功，不删除任务」每秒重试一次，45 秒都跑不完，最后是我手动停的（上面截的是前 14 行，之后同一组三行每秒重复一次）。

注意那条 SQL：insert into \fp31:seeds` (`state`, `id`) … on duplicate key update` —— 它把 redis 的 key 名当表名去更新了，所以「redis 任务表 + update_task_batch」是互斥组合，框架不会拦，只会一直重试。

MySQL 做任务表：表结构与 state 语义

用 MySQL 做种子表（TaskSpider 和 BatchSpider 的标准用法）需要先建好任务表，官方给的建表语句：

1
2
3
4
5
6
CREATE TABLE `batch_spider_task` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `url` varchar(255) DEFAULT NULL,
  `state` int(11) DEFAULT '0',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

task_state 字段（默认名 state）有四个值，框架靠它推进度：

值 含义 谁来改
0 待抓取 框架下发时批量置 2；每批开始还会把非 -1 的重置为 0
1 抓取完毕 自己在 parse 里 yield self.update_task_batch(request.task_id)
2 抓取中 框架下发时写
-1 抓取失败/无效 自己写；-1 永远不会再被抓

流程是：master 分批把 state=0 的任务下发到 redis 并置 2 → worker 消费 → 代码里置 1 → 队列空了 master 再查表里是否还有 2（视为丢失，重置为 0 再下发），直到表里只剩 1 和 -1。

BatchSpider 在这之上多了 batch_record_table（批次记录表，框架自动建）、batch_name、batch_interval 三个参数，采集时会给每条数据带上 batch_date，本批次没跑完下一批次不会开始。官方给的审核 SQL 就是一句 SELECT count(1), batch_date FROM 表名 GROUP BY batch_date;。

下面这段把「master 下发 → worker 消费 → worker 被硬杀 → 积压」四个阶段串起来跑，每阶段打印任务表的 state 分布和 redis 队列（可见 指 score <= now 的条数）：

task_mysql.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
# -*- coding: utf-8 -*-
"""TaskSpider 用 MySQL 做任务表：state 的真实流转（0 待做 / 2 抓取中 / 1 做完）与硬杀 worker 后的积压。

    python task_mysql.py master   # 只下发任务（阻塞）
    python task_mysql.py worker   # 只消费任务
    python task_mysql.py          # 串起整轮实验
"""

import os
import subprocess
import sys
import time

import pymysql
import redis
import feapder

PW = os.environ["MYSQL_PW"]                     # 口令走环境变量，不要把明文写进代码
DB = "feapder_probe"
TASK_TABLE = "movie_task"
RK = "fp40:movie_task"
URLS = ["https://ssr1.scrape.center/detail/%d" % i for i in range(1, 21)]


def mysql():
    return pymysql.connect(host="127.0.0.1", port=3306, user="root", password=PW,
                           database=DB, charset="utf8mb4", connect_timeout=8, autocommit=True)


def init():
    with mysql() as c, c.cursor() as cur:
        cur.execute("drop table if exists `%s`" % TASK_TABLE)
        cur.execute("create table `%s` (`id` int unsigned not null auto_increment primary"
                    " key, `url` varchar(255) default null, `state` int default 0,"
                    " unique key `uk_url` (`url`)) engine=InnoDB default charset utf8mb4"
                    % TASK_TABLE)
        sql = "insert into `%s` (`url`, `state`) values (%%s, 0)" % TASK_TABLE
        for u in URLS:
            cur.execute(sql, [u])
    print("灌 %d 条任务，state 全 0" % len(URLS))


def state(label):
    with mysql() as c, c.cursor() as cur:
        cur.execute("select state, count(*) from `%s` group by state order by state" % TASK_TABLE)
        rows = cur.fetchall()
    cli = redis.Redis(host="127.0.0.1", port=6390)
    key = RK + ":z_requests"
    now = time.time()
    total = cli.zcard(key)
    visible = cli.zcount(key, "-inf", now)
    print("%s 表里 %s | redis 队列 %d 条（可见 %d）" % (label, rows, total, visible))


class MovieTask(feapder.TaskSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0,
        MYSQL_IP="127.0.0.1", MYSQL_PORT=3306, MYSQL_DB=DB,
        MYSQL_USER_NAME="root", MYSQL_USER_PASS=PW,
        ITEM_PIPELINES=["feapder.pipelines.console_pipeline.ConsolePipeline"])

    def start_requests(self, task):
        yield feapder.Request(task["url"], task_id=task["id"])

    def parse(self, request, response):
        item = feapder.Item()
        item.table_name = "movie_task_result"
        item.title = response.css("h2::text").extract_first()
        yield item
        yield self.update_task_batch(request.task_id)   # 做完一条，state 置 1


def build():
    return MovieTask(redis_key=RK, task_table=TASK_TABLE, task_table_type="mysql",
                     task_keys=["id", "url"], min_task_count=1, check_task_interval=2,
                     task_limit=3, thread_count=1)


def child(role, seconds, log):
    p = subprocess.Popen([sys.executable, "-X", "utf8", "-u", __file__, role],
                         stdout=log, stderr=subprocess.STDOUT, text=True)
    time.sleep(seconds)
    p.kill()
    time.sleep(0.5)


if len(sys.argv) > 1:
    spider = build()
    if sys.argv[1] == "master":
        spider.start_monitor_task()      # 只下发
    else:
        spider.start()                   # 只消费


else:
    cli = redis.Redis(host="127.0.0.1", port=6390)
    for k in list(cli.keys(RK + "*")):
        cli.delete(k)
    init()
    state("初始")

    log = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "phase.log"),
               "w", encoding="utf-8")

    print("\n== 1) 只起 master 6 秒：按 task_limit=3 下发，state 0 -> 2")
    child("master", 6, log)
    state("下发后")

    print("\n== 2) 只起 worker 12 秒：消费 redis 里的 3 条，parse 里置 state=1")
    child("worker", 12, log)
    state("消费后")

    print("\n== 3) 重灌 20 条，master 常驻 + worker 跑 12 秒被硬杀")
    init()
    m = subprocess.Popen([sys.executable, "-X", "utf8", "-u", __file__, "master"],
                         stdout=log, stderr=subprocess.STDOUT, text=True)
    time.sleep(3)
    child("worker", 12, log)
    state("硬杀后")

    print("\n== 4) 什么都不做再等 8 秒：master 只说「有积压任务」，既不补发也不重置")
    time.sleep(8)
    state("又等 8 秒")
    m.kill()
    log.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
灌 20 条任务，state 全 0
初始 表里 ((0, 20),) | redis 队列 0 条（可见 0）

== 1) 只起 master 6 秒：按 task_limit=3 下发，state 0 -> 2
下发后 表里 ((0, 17), (2, 3)) | redis 队列 3 条（可见 3）

== 2) 只起 worker 12 秒：消费 redis 里的 3 条，parse 里置 state=1
消费后 表里 ((0, 17), (1, 3)) | redis 队列 0 条（可见 0）

== 3) 重灌 20 条，master 常驻 + worker 跑 12 秒被硬杀
灌 20 条任务，state 全 0
硬杀后 表里 ((0, 8), (1, 9), (2, 3)) | redis 队列 3 条（可见 0）

== 4) 什么都不做再等 8 秒：master 只说「有积压任务」，既不补发也不重置
又等 8 秒 表里 ((0, 8), (1, 9), (2, 3)) | redis 队列 3 条（可见 0）
  • state 的流转全对得上源码：master 下发时 update movie_task set state = 2 where id in (...)，worker 在 parse 里 yield self.update_task_batch(request.task_id) 把它置 1。
  • 第 3 段那两个数字取决于硬杀那一刻单线程跑到第几条，重跑会看到 ((0, 11), (1, 7), (2, 2)) | redis 队列 2 条（可见 0） 这类分布。要看的是三件事同时成立：表里有 2、队列里有条数、可见 是 0。前两段（只起 master、只起 worker）是稳定的。
  • 硬杀 worker 那一刻，手里那条任务在 redis 里的 score 已被改到 now + 600（REQUEST_LOST_TIMEOUT=600），所以队列里还有 3 条、可见 0 条；master 每 2 秒巡检一次，只打「redis 中尚有3条积压任务，暂时不派发新任务」——它按 zcard 的总数判断积压，看不见「取出中」的任务，于是既不补发也不执行 reset_lose_task_from_mysql。干等 8 秒，两个数字一点没变。
  • 恢复办法有三条：等 600 秒超时自动可见；重启时给 delete_keys 传 "*z_requests" 清空队列（配合 update movie_task set state = 0 where state = 2 手工补刀最稳）；或者把 REQUEST_LOST_TIMEOUT 调小。

update_task_batch 会被非空列卡死

update_task_batch 返回的是一个只带 {id, state} 的 UpdateItem，落库走的是 insert into 任务表 (id, state) values (...) on duplicate key update ...——它要往任务表插一行只填了两个字段的数据，靠唯一索引冲突转成更新。任务表里只要还有别的 NOT NULL 列，这条 SQL 就失败：

task_mysql_str.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
# -*- coding: utf-8 -*-
"""任务表里有 NOT NULL 列时，update_task_batch 与 update_task_state 的差别。

    python task_mysql_str.py batch    # yield self.update_task_batch(task_id)
    python task_mysql_str.py single   # yield lambda: self.update_task_state(task_id)
    python task_mysql_str.py          # 两种写法各跑一遍，打印 state 分布
"""

import os
import subprocess
import sys
import time

import pymysql
import redis
import feapder

PW = os.environ["MYSQL_PW"]   # 口令走环境变量，不要把明文写进代码
DB = "feapder_probe"
TASK_TABLE = "movie_task3"
RK = "fp42:movie_task3"
URLS = ["https://ssr1.scrape.center/detail/%d" % i for i in range(1, 5)]
LOG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "task_mysql_str.log")


def mysql():
    return pymysql.connect(host="127.0.0.1", port=3306, user="root", password=PW,
                           database=DB, charset="utf8mb4", connect_timeout=8, autocommit=True)


def init():
    with mysql() as c, c.cursor() as cur:
        cur.execute("drop table if exists `%s`" % TASK_TABLE)
        cur.execute("create table `%s` (`id` int unsigned not null auto_increment primary"
                    " key, `url` varchar(255) not null, `state` int default 0,"
                    " unique key `uk_url` (`url`)) engine=InnoDB default charset utf8mb4"
                    % TASK_TABLE)
        sql = "insert into `%s` (`url`, `state`) values (%%s, 0)" % TASK_TABLE
        for u in URLS:
            cur.execute(sql, [u])
    print("灌 %d 条任务（url 列 NOT NULL）" % len(URLS))


def state(label):
    with mysql() as c, c.cursor() as cur:
        cur.execute("select state, count(*) from `%s` group by state order by state" % TASK_TABLE)
        print("%s 表里 %s" % (label, cur.fetchall()))


class StrTask(feapder.TaskSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0,
        MYSQL_IP="127.0.0.1", MYSQL_PORT=3306, MYSQL_DB=DB,
        MYSQL_USER_NAME="root", MYSQL_USER_PASS=PW,
        ITEM_PIPELINES=["feapder.pipelines.console_pipeline.ConsolePipeline"])
    way = "batch"

    def start_requests(self, task):
        yield feapder.Request(task["url"], task_id=task["id"])

    def parse(self, request, response):
        item = feapder.Item()
        item.table_name = "movie3_result"
        item.title = response.css("h2::text").extract_first()
        yield item
        if self.way == "batch":
            yield self.update_task_batch(request.task_id)
        else:
            yield lambda: self.update_task_state(request.task_id)


def build():
    return StrTask(redis_key=RK, task_table=TASK_TABLE, task_table_type="mysql",
                   task_keys=["id", "url"], min_task_count=1, check_task_interval=2,
                   task_limit=4, thread_count=1)


if len(sys.argv) > 1:
    role = sys.argv[1]
    if role == "master":
        build().start_monitor_task()
    else:
        StrTask.way = sys.argv[2]
        build().start()
else:
    open(LOG, "w").close()
    cli = redis.Redis(host="127.0.0.1", port=6390)
    fh = open(LOG, "a", encoding="utf-8")

    for way in ("batch", "single"):
        for k in list(cli.keys(RK + "*")):
            cli.delete(k)
        init()
        print("\n== 写法：%s" % way)
        m = subprocess.Popen([sys.executable, "-X", "utf8", "-u", __file__, "master"],
                             stdout=fh, stderr=subprocess.STDOUT, text=True)
        time.sleep(3)
        w = subprocess.Popen([sys.executable, "-X", "utf8", "-u", __file__, "worker", way],
                             stdout=fh, stderr=subprocess.STDOUT, text=True)
        time.sleep(18)
        w.kill()
        m.kill()
        state("%s 跑完" % way)
    fh.close()
    print("（框架日志在 task_mysql_str.log）")
1
2
3
4
5
6
7
8
9
灌 4 条任务（url 列 NOT NULL）

== 写法：batch
batch 跑完 表里 ((2, 4),)
灌 4 条任务（url 列 NOT NULL）

== 写法：single
single 跑完 表里 ((1, 4),)
（框架日志在 task_mysql_str.log）
1
2
Thread-3|2026-11-07 23:16:43,585|item_buffer.py|__export_to_db|line:278|ERROR| MysqlPipeline 更新数据失败. table: movie_task3  items: [{'id': 1, 'state': 1}]
Thread-3|2026-11-07 23:16:43,589|item_buffer.py|__add_item_to_db|line:396|ERROR| 入库不成功，不删除任务，将自动重试，失败items:
  • 上面第 1 轮（batch）4 条任务全卡在 state=2，永远做不完；第 2 轮换成 yield lambda: self.update_task_state(request.task_id) 就正常置 1，因为这条走的是 update 表 set state=1 where id = N，不插行。
  • 官方示例建表用的是 param varchar(1000) DEFAULT NULL，可空，所以照抄官方语句不会踩到；自己加 url varchar(255) NOT NULL 才会炸。要么把非空列改成可空，要么改用 update_task_state。

BatchSpider 的批次记录表

batch_record_table 是框架自己建的（create_batch_record_table()），批次进度看它，不看任务表。batch_interval=0 时批次时间精确到分钟（源码里 %Y-%m-%d %H:%M），跨一分钟就换新批次：

batch_mysql.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
# -*- coding: utf-8 -*-
"""BatchSpider + MySQL：批次记录表是否框架自动建、batch_date 怎么落到数据上。

    python batch_mysql.py master / worker   # 两个角色
    python batch_mysql.py                   # 串起实验
"""

import os
import subprocess
import sys
import time

import pymysql
import redis
import feapder
from feapder.core.spiders.batch_spider import BatchSpider

PW = os.environ["MYSQL_PW"]   # 口令走环境变量，不要把明文写进代码
DB = "feapder_probe"
TASK_TABLE = "movie_batch_task"
BATCH_TABLE = "movie_batch_record"
RESULT = "movie_batch_result"
RK = "fp43:movie_batch"
URLS = ["https://ssr1.scrape.center/detail/%d" % i for i in range(1, 6)]


def mysql():
    return pymysql.connect(host="127.0.0.1", port=3306, user="root", password=PW,
                           database=DB, charset="utf8mb4", connect_timeout=8, autocommit=True)


def q(text, sql):
    with mysql() as c, c.cursor() as cur:
        cur.execute(sql)
        print("%s ->" % text, cur.fetchall())


def init():
    with mysql() as c, c.cursor() as cur:
        cur.execute("drop table if exists `%s`" % TASK_TABLE)
        cur.execute("drop table if exists `%s`" % RESULT)
        cur.execute("drop table if exists `%s`" % BATCH_TABLE)
        cur.execute("create table `%s` (`id` int unsigned not null auto_increment primary"
                    " key, `url` varchar(255) default null, `state` int default 0,"
                    " unique key `uk_url` (`url`)) engine=InnoDB default charset utf8mb4"
                    % TASK_TABLE)
        # 结果表：batch_date 这一列要自己写（框架不会自动加进 item）
        cur.execute("create table `%s` (`id` int unsigned not null auto_increment primary"
                    " key, `title` varchar(255) default null, `score` decimal(3,1) default"
                    " null, `batch_date` date default null,"
                    " unique key `uk_title` (`title`)) engine=InnoDB default charset utf8mb4"
                    % RESULT)
        sql = "insert into `%s` (`url`, `state`) values (%%s, 0)" % TASK_TABLE
        for u in URLS:
            cur.execute(sql, [u])
    print("任务表灌 %d 条；结果表建好（含 batch_date 列）" % len(URLS))


class MovieBatch(BatchSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        REDISDB_IP_PORTS="127.0.0.1:6390", REDISDB_USER_PASS="", REDISDB_DB=0,
        MYSQL_IP="127.0.0.1", MYSQL_PORT=3306, MYSQL_DB=DB,
        MYSQL_USER_NAME="root", MYSQL_USER_PASS=PW,
        ITEM_PIPELINES=["feapder.pipelines.mysql_pipeline.MysqlPipeline"])
    stamp = False      # 是否自己把批次时间写进 item

    def start_requests(self, task):
        yield feapder.Request(task["url"], task_id=task["id"])

    def parse(self, request, response):
        item = feapder.Item()
        item.table_name = RESULT
        item.title = response.css("h2::text").extract_first()
        item.score = float(response.css("p.score::text").extract_first())
        if self.stamp:
            item.batch_date = self.batch_date   # 框架只给属性，不自动塞进 item
        yield item
        yield self.update_task_batch(request.task_id)


def build():
    return MovieBatch(redis_key=RK, task_table=TASK_TABLE, batch_record_table=BATCH_TABLE,
                      batch_name="电影详情批次", batch_interval=0, task_table_type="mysql",
                      task_keys=["id", "url"], min_task_count=1, check_task_interval=2,
                      thread_count=2)


if len(sys.argv) > 1:
    MovieBatch.stamp = sys.argv[1] == "stamp"
    spider = build()
    if os.environ.get("ROLE") == "master":
        spider.start_monitor_task()
    else:
        spider.start()
else:
    log = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "batch_mysql.log"),
               "w", encoding="utf-8")
    cli = redis.Redis(host="127.0.0.1", port=6390)

    def clean_keys():
        for k in list(cli.keys(RK + "*")):
            cli.delete(k)

    def run_round(stamp):
        os.environ["ROLE"] = "master"
        m = subprocess.Popen([sys.executable, "-X", "utf8", "-u", __file__,
                              "stamp" if stamp else "nostamp"],
                             stdout=log, stderr=subprocess.STDOUT, text=True)
        time.sleep(3)
        os.environ["ROLE"] = "worker"
        w = subprocess.Popen([sys.executable, "-X", "utf8", "-u", __file__,
                              "stamp" if stamp else "nostamp"],
                             stdout=log, stderr=subprocess.STDOUT, text=True)
        time.sleep(22)
        w.kill()
        m.kill()

    clean_keys()
    init()
    q("跑之前 show tables", "show tables like 'movie_batch%%'")

    print("\n== 第 1 轮：parse 里不写 batch_date")
    run_round(False)
    q("show tables", "show tables like 'movie_batch%%'")
    q("批次记录表（框架自动建的）", "select batch_date, total_count, done_count, fail_count,"
                                  " is_done from `%s` order by id" % BATCH_TABLE)
    q("结果表", "select title, score, batch_date from `%s` order by id" % RESULT)

    print("\n== 第 2 轮：parse 里写 item.batch_date = self.batch_date")
    with mysql() as c, c.cursor() as cur:
        cur.execute("truncate table `%s`" % RESULT)
        cur.execute("update `%s` set state = 0" % TASK_TABLE)
    clean_keys()
    run_round(True)
    q("结果表", "select title, score, batch_date from `%s` order by id" % RESULT)
    q("批次记录表", "select batch_date, total_count, done_count, is_done from `%s` order by id"
                    % BATCH_TABLE)
    q("任务表 state", "select state, count(*) from `%s` group by state" % TASK_TABLE)
    log.close()
1
2
3
4
5
6
7
8
9
10
11
12
任务表灌 5 条；结果表建好（含 batch_date 列）
跑之前 show tables -> (('movie_batch_result',), ('movie_batch_task',))

== 第 1 轮：parse 里不写 batch_date
show tables -> (('movie_batch_record',), ('movie_batch_result',), ('movie_batch_task',))
批次记录表（框架自动建的） -> ((datetime.datetime(2026, 11, 7, 23, 22), 5, 5, 0, 1), (datetime.datetime(2026, 11, 7, 23, 22), 5, 0, None, 0))
结果表 -> (('这个杀手不太冷 - Léon', Decimal('9.5'), None), ('霸王别姬 - Farewell My Concubine', Decimal('9.5'), None), ('肖申克的救赎 - The Shawshank Redemption', Decimal('9.5'), None), ('泰坦尼克号 - Titanic', Decimal('9.5'), None), ('罗马假日 - Roman Holiday', Decimal('9.5'), None))

== 第 2 轮：parse 里写 item.batch_date = self.batch_date
结果表 -> (('霸王别姬 - Farewell My Concubine', Decimal('9.5'), datetime.date(2026, 11, 7)), ('这个杀手不太冷 - Léon', Decimal('9.5'), datetime.date(2026, 11, 7)), ('肖申克的救赎 - The Shawshank Redemption', Decimal('9.5'), datetime.date(2026, 11, 7)), ('泰坦尼克号 - Titanic', Decimal('9.5'), datetime.date(2026, 11, 7)), ('罗马假日 - Roman Holiday', Decimal('9.5'), datetime.date(2026, 11, 7)))
批次记录表 -> ((datetime.datetime(2026, 11, 7, 23, 22), 5, 5, 1), (datetime.datetime(2026, 11, 7, 23, 22), 5, 5, 1), (datetime.datetime(2026, 11, 7, 23, 22), 5, 0, 0))
任务表 state -> ((0, 5),)
  • 跑之前只有任务表和结果表，跑完多出 movie_batch_record，字段是 batch_date / total_count / done_count / fail_count / interval / interval_unit / create_time / update_time / is_done，没有 batch_name 列——batch_name 只是程序里的标签，不落库。
  • 第 1 轮 parse 里没写 item.batch_date，结果表那一列全是 None：框架不会自动把批次时间塞进 item，它只暴露 self.batch_date 属性（同时写进环境变量 batch_date）。第 2 轮显式赋值之后才落进去。
  • 每批开始框架会执行 update 任务表 set state = 0 where state != -1（源码 init_task()），所以跑完一轮再等新批次，state 会被清回 0；判断做完没做完请看批次表的 done_count / is_done。
  • 官方那句审核 SQL 可以直接用：SELECT count(1), batch_date FROM 结果表 GROUP BY batch_date;。

浏览器渲染

SPA 页面直接请求拿不到数据。feapder 的做法是在 Request 上挂 render=True，交给渲染下载器：

render_spa.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
import feapder

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

WEBDRIVER = dict(
    pool_size=1, load_images=False, headless=True, driver_type="CHROME",
    timeout=30, window_size=(1024, 800), executable_path=None,
    render_time=3, auto_install_driver=True,
    custom_argument=["--ignore-certificate-errors"],
)


class RenderSpider(feapder.AirSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        WEBDRIVER=WEBDRIVER,
    )

    def start_requests(self):
        yield feapder.Request(SPA)                       # 普通下载
        yield feapder.Request(SPA, render=True, callback=self.parse_render)
        yield feapder.Request(SPA, render=True, render_time=0, callback=self.parse_zero)

    def parse(self, request, response):
        print("[普通请求] 字节 %d | 有片名? %s | 卡片数 %d" % (
            len(response.text), "霸王别姬" in response.text, len(response.css("div.el-card"))))

    def parse_render(self, request, response):
        print("[render + render_time=3] 字节 %d | 有片名? %s | 卡片数 %d" % (
            len(response.text), "霸王别姬" in response.text, len(response.css("div.el-card"))))

    def parse_zero(self, request, response):
        print("[render + render_time=0] 字节 %d | 有片名? %s | 卡片数 %d" % (
            len(response.text), "霸王别姬" in response.text, len(response.css("div.el-card"))))


if __name__ == "__main__":
    RenderSpider(thread_count=1).start()
1
2
3
[普通请求] 字节 952 | 有片名? False | 卡片数 0
[render + render_time=3] 字节 27911 | 有片名? True | 卡片数 10
[render + render_time=0] 字节 2934 | 有片名? False | 卡片数 0

把 render_time 单独拎出来测才看得清。同一个浏览器实例、同一个 SPA 页面，取两次源码：一次 get 完立刻取，一次等 3 秒再取。

render_race.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import time

from feapder.utils.webdriver import SeleniumDriver

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

driver = SeleniumDriver(headless=True, load_images=False, timeout=30,
                        auto_install_driver=True, driver_type="CHROME",
                        window_size=(1024, 800))
for i in range(3):
    t0 = time.time()
    driver.get(SPA)
    fast = driver.page_source                      # 加载完立刻取
    time.sleep(3)
    slow = driver.page_source                      # 等 3 秒再取
    print("第%d轮: get 耗时 %.1fs | 立刻取 %d 字节, el-card %d 次 | 等3秒后 %d 字节, el-card %d 次" % (
        i + 1, time.time() - t0, len(fast), fast.count("el-card"),
        len(slow), slow.count("el-card")))
driver.quit()
1
2
3
第1轮: get 耗时 0.8s | 立刻取 2490 字节, el-card 0 次 | 等3秒后 27911 字节, el-card 20 次
第2轮: get 耗时 0.0s | 立刻取 2541 字节, el-card 0 次 | 等3秒后 27911 字节, el-card 20 次
第3轮: get 耗时 0.0s | 立刻取 2544 字节, el-card 0 次 | 等3秒后 27911 字节, el-card 20 次

driver.get() 返回只代表文档加载完，卡片是随后由 JS 打接口填进去的，所以「立刻取」拿到的 2.5 KB 里没有一张卡；render_time 就是这段等待——源码里 SeleniumDownloader.download 是「browser.get(url) → render_time 非 0 才 delay_time(render_time) → 取 page_source」。

于是默认值 render_time=0 的含义是「一点不等」，能不能抓到数据纯看接口回来的快慢：上面那三轮里立刻取全部为空，而上一节 render_spa.py 那次运行碰巧 0 秒也拿到了 28821 字节。要稳，就把 render_time 给够（这站 3 秒稳定，2 秒时好时坏），或者直接走接口 spa1.scrape.center/api/movie?limit=10&offset=0——更快也更稳，浏览器渲染留给没有接口的页面。

用浏览器渲染前要知道的几件事：

render_details.py
1
2
3
4
5
6
7
8
9
10
11
12
13
import time

from feapder import Request, setting
from feapder.network.downloader._selenium import SeleniumDownloader

setting.WEBDRIVER = dict(pool_size=1, load_images=False, headless=True,
                         driver_type="CHROME", timeout=30, window_size=(1024, 800),
                         auto_install_driver=True, render_time=3)
dl = SeleniumDownloader()
resp = dl.download(Request("https://spa1.scrape.center/not-exist-page", render=True))
print("渲染模式 status_code =", resp.status_code, "| elapsed =", resp.elapsed)
print("url =", resp.url, "| 字节", len(resp.text))
print("响应头里 feapder 自己拼的:", sorted(resp.headers.keys()))
1
2
3
渲染模式 status_code = 200 | elapsed = 666
url = https://spa1.scrape.center/not-exist-page | 字节 1265
响应头里 feapder 自己拼的: ['Cookie', 'User-Agent']
  • 渲染模式的 status_code 恒为 200，elapsed 恒等于 666 微秒（源码把这两个值写死在交给 Response.from_dict 的字典里），所以 404/500 在渲染下看不出来，只能靠正文特征或 validate 判断。
  • 响应头不是服务器给的那套，只有框架用 browser.current_url、browser.cookies、browser.user_agent 拼出来的 User-Agent 和 Cookie。
  • 驱动从哪来：auto_install_driver=True 时用 webdriver_manager 下载匹配本机 Chrome 的 chromedriver，缓存到用户目录的 .wdm 下（实测第一次跑要几十秒下载，之后 3~10 秒一轮）；也可以 executable_path 指定现成的驱动，写法与《爬虫-Selenium和PhantomJS》里的 Service 一致。
  • 依赖必须齐：只装 selenium 不够，feapder.utils.webdriver 会把 Playwright 驱动一起 import；换成 Playwright 只需 RENDER_DOWNLOADER = "feapder.network.downloader.PlaywrightDownloader"，它多一个 wait_until="networkidle" 参数（比 selenium 的固定 sleep 更接近「等到数据出现」）。
  • 文档里的 driver_type 还支持 EDGE、FIREFOX、PHANTOMJS；PhantomJS 早已停止维护，Selenium 4 里也删了对应入口，实际只有 CHROME/EDGE/FIREFOX 可用（详见《爬虫-Selenium和PhantomJS》）。
  • WEBDRIVER 里还有 xhr_url_regexes（拦截 XHR 接口直接拿 JSON）、use_stealth_js、proxy、custom_argument 等参数；代理优先级是「Request 指定 > 配置文件 PROXY_EXTRACT_API > WEBDRIVER 里的 proxy」。

代理与 UA

三种用法：内置代理池（配提取 API）、自定义代理池、请求里直接指定。先把内置池的行为看清——它对返回正文的要求非常苛刻。

proxy_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
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

from feapder.network.proxy_pool import ProxyPool

BODY = {
    "/crlf": "127.0.0.1:8011\r\n127.0.0.1:8011\r\n127.0.0.1:9999\r\n",
    "/lf": "127.0.0.1:8011\n127.0.0.1:8011\n",
    "/json": '{"code":0,"data":[{"proxy":"127.0.0.1:8011"}]}',
    "/empty": "",
}


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = BODY.get(self.path, "").encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


srv = HTTPServer(("127.0.0.1", 8023), Handler)
threading.Thread(target=srv.serve_forever, daemon=True).start()

API = "http://127.0.0.1:8023%s"
for path in ("/crlf", "/lf", "/json", "/empty"):
    pool = ProxyPool(proxy_api=API % path)
    try:
        print("API %-8s pull_proxies -> %r" % (path, pool.pull_proxies()))
        print("          get_proxy   -> %r" % (pool.get_proxy(),))
    except Exception as exc:
        print("API %-8s -> %s: %s" % (path, type(exc).__name__, str(exc)[:90]))
srv.shutdown()
1
2
3
4
5
6
API /crlf    pull_proxies -> ['127.0.0.1:8011', '127.0.0.1:8011', '127.0.0.1:9999']
          get_proxy   -> {'http': 'http://127.0.0.1:8011', 'https': 'http://127.0.0.1:8011'}
API /lf      pull_proxies -> ['127.0.0.1:8011\n127.0.0.1:8011']
          get_proxy   -> {'http': 'http://127.0.0.1:8011\n127.0.0.1:8011', 'https': 'http://127.0.0.1:8011\n127.0.0.1:8011'}
API /json    pull_proxies -> Exception: ('获取代理失败', '{"code":0,"data":[{"proxy":"127.0.0.1:8011"}]}')
API /empty   pull_proxies -> Exception: ('获取代理失败', '')
  • 分隔符只认 \r\n：正文用 \n 分隔时不报错，而是把整篇当一个代理用，请求必然失败——这类错误最难查。
  • 判定「失败」的规则只有一条：正文为空或含 {（认为是接口返回的 JSON 而非纯代理列表）就抛 获取代理失败，外面套 @tools.retry(3, interval=5) 重试 3 次、每次间隔 5 秒。代理站返回 HTML 欢迎页/错误页时会被当成代理硬用。
  • 交给 requests 的字典是 {"http": "http://ip:port", "https": "http://http://..."}——https 那一档也用 http:// 前缀，是 format_proxy 的固定写法。

「真的走了代理」要有证据才敢下结论。这段起一个回显服务（把收到的请求头原样打回来），代理 API 指向本机 127.0.0.1:8011 那个 HTTP 代理——只要请求经过它，回显里就会多出 Via 与 X-Forwarded-For：

proxy_evidence.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
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import feapder
from feapder import setting

HITS = []


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path.startswith("/api"):
            body = b"127.0.0.1:8011\r\n127.0.0.1:8011\r\n"
            head = "text/plain"
        else:                      # /echo：把收到的请求头原样返回
            body = ("\n".join("%s: %s" % (k, v) for k, v in self.headers.items())).encode()
            head = "text/plain; charset=utf-8"
        self.send_response(200)
        self.send_header("Content-Type", head)
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


srv = HTTPServer(("127.0.0.1", 8022), Handler)
threading.Thread(target=srv.serve_forever, daemon=True).start()

SEEN = []


class ProxySpider(feapder.AirSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        PROXY_ENABLE=True, PROXY_EXTRACT_API="http://127.0.0.1:8022/api")

    def start_requests(self):
        yield feapder.Request("http://127.0.0.1:8022/echo?i=0")
        yield feapder.Request("http://127.0.0.1:8022/echo?i=1")

    def parse(self, request, response):
        via = [ln for ln in response.text.splitlines()
               if ln.lower().startswith(("via", "x-forwarded-for", "host"))]
        SEEN.append((request.url, request.get_proxy(), via))


ProxySpider(thread_count=1).start()
for url, proxy, via in SEEN:
    print("请求 %s | 用的代理 %s | 回显头 %s" % (url, proxy, via))

setting.PROXY_ENABLE = False       # 关掉代理再问一次做对照
r = feapder.Request("http://127.0.0.1:8022/echo?i=2")
resp = r.get_response()
print("不走代理 | 用的代理 %r | 回显头 %s" % (
    r.get_proxy(), [ln for ln in resp.text.splitlines()
                    if ln.lower().startswith(("via", "x-forwarded-for"))]))
srv.shutdown()
1
2
3
4
ProxySpider|2026-11-07 21:40:58,640|air_spider.py|run|line:114|INFO| 无任务，爬虫结束
请求 http://127.0.0.1:8022/echo?i=0 | 用的代理 127.0.0.1:8011 | 回显头 ['Host: 127.0.0.1:8022', 'X-Forwarded-For: 127.0.0.1', 'Via: 1.1 spider-proxy']
请求 http://127.0.0.1:8022/echo?i=1 | 用的代理 127.0.0.1:8011 | 回显头 ['Host: 127.0.0.1:8022', 'X-Forwarded-For: 127.0.0.1', 'Via: 1.1 spider-proxy']
不走代理 | 用的代理 None | 回显头 []
  • 文档说「自动随机使用上面的代理」，实际 get_proxy 是 Queue.get_nowait() 后放回队尾，是轮转不是随机；一次 API 拉回来的 N 条会依次用完，队列空了才再调一次 API。
  • 上例第 5 个请求撞进死代理 127.0.0.1:9999，抛 ProxyError，然后 del_proxy 把它删掉，重试时重新取到可用代理并成功。删代理的时机在源码里是 request.retry_times % PROXY_MAX_FAILED_TIMES == 0（默认 5），也就是「第 5、10 次重试才删」，而且只有下载类异常才触发。
  • request.del_proxy() 会连带把当前请求的 proxies 删掉，所以紧接着 request.get_proxy() 是 None——不是没删干净，是这条请求已经不带代理了。
  • 内置池没有黑名单：如果 API 每次返回同一批死代理，就会形成「拉取→失败→删除→再拉同一批」的死循环（实测每 4 秒命中一次 API，del_proxy 被调 30 次），直到重试次数用完。

代码里的三种写法：

proxy_usage.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
import feapder


class ProxySpider(feapder.AirSpider):
    __custom_setting__ = dict(
        LOG_LEVEL="INFO", LOG_COLOR=False, PRINT_EXCEPTION_DETAILS=False,
        PROXY_ENABLE=True,
        PROXY_EXTRACT_API="http://127.0.0.1:8023/crlf",   # 正文必须是 \r\n 分隔
    )

    def start_requests(self):
        yield feapder.Request("https://httpbin.org/ip")                    # 走代理池
        yield feapder.Request("https://httpbin.org/ip", proxies=-1)         # 同上（-1 表示按配置取）
        yield feapder.Request("https://httpbin.org/ip", proxies={           # 直接指定，跳过代理池
            "http": "http://127.0.0.1:8011", "https": "http://127.0.0.1:8011"})

    def parse(self, request, response):
        print("origin =", response.json.get("origin"),
              "| custom_proxies =", request.custom_proxies)

    def exception_request(self, request, response):
        request.del_proxy()      # 出异常就把这个代理还给池子删掉


if __name__ == "__main__":
    ProxySpider(thread_count=1).start()

PROXY_ENABLE=False 时代理 API 一次都不会被调用，全部请求直连；反过来，PROXY_ENABLE=True 但 PROXY_EXTRACT_API 为空也不会走代理——判断条件是 if proxies == -1 and setting.PROXY_ENABLE and setting.PROXY_EXTRACT_API。

自定义代理池只要继承 BaseProxyPool 并实现 get_proxy/del_proxy，然后在配置里写模块路径字符串（一个进程只实例化一次）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# my_proxypool.py
from feapder.network.proxy_pool import BaseProxyPool


class MyProxyPool(BaseProxyPool):
    def get_proxy(self):
        """返回 {"http": "http://ip:port", "https": "http://ip:port"} 或 None"""
        return {"http": "http://127.0.0.1:8011", "https": "http://127.0.0.1:8011"}

    def del_proxy(self, proxy):
        """proxy 是不带 http:// 前缀的 ip:port"""
        ...

# 爬虫里
__custom_setting__ = dict(PROXY_POOL="my_proxypool.MyProxyPool", PROXY_ENABLE=True)

实测两个细节：PROXY_POOL 的路径是相对当前工作目录的模块名（跟 setting.py 一个道理，脚本换目录跑就 import 不到）；get_proxy() 返回 None 时框架是 while True 死循环打 暂无可用代理，池子空了要返回一个真值或让爬虫别用它。

公开代理站能不能直接当 API？https://proxypool.scrape.center/random 返回纯文本 ip:port（本机这次取到 186.10.150.185:8080，19 字节，不是固定的 13 字节），切分后正好一个代理，实测确实抓到过 ssr1 首页 10 张卡：

1
[PUB] 成功 url=https://ssr1.scrape.center/ 代理='144.124.251.24:10605' 状态码=200 卡片10张

但这类站「取到」不等于「能用」，连不上时报的是 ProxyError('Unable to connect to proxy', ConnectTimeoutError(... 'Connection to 1.1.1.1 timed out. (connect timeout=22)'))，一次失败要等满 22 秒，默认重试 10 次意味着一个坏代理能吃掉三四分钟——代理相关写法与更多坑见《爬虫-代理》。

UserAgent 与用户池

随机 UA 由 RANDOM_HEADERS + USER_AGENT_TYPE（chrome/opera/firefox/internetexplorer/safari/mobile）+ DEFAULT_USERAGENT 三个键控制，行为见「Request 参数」。

feapder.network.user_pool 里的 GuestUserPool / NormalUserPool / GoldUserPool 是给「需要登录态的大规模采集」准备的账号池，实测三条要知道的结论：

  • 没有 feapder.user_pool 这个顶层模块，入口只有 from feapder.network.user_pool import GuestUserPool, GuestUser, ...。
  • 框架不会自动把用户池接到爬虫上：整个 feapder 包里除了 network/user_pool/ 自己，没有任何文件引用它，cookie/代理要自己在 download_midware 里塞。
  • GuestUserPool 只要 redis（键 fpD:user_pool:h_guest_pool，来自模板 {redis_key}:h_{user_type}_pool），但默认 login() 走浏览器渲染，要重写；NormalUserPool 需要 MySQL；GoldUserPool 不重写 login() 时 get_user() 会 TypeError: eval() arg 1 must be a string, bytes or code object 无限刷屏、进程不退出。

报警与监控

报警相关键共四个渠道（钉钉 / 飞书 / 企业微信 / 邮件）加三个开关，实测默认值：

warning_keys.py
1
2
3
4
5
6
7
8
9
from feapder import setting

for k in ("DINGDING_WARNING_URL", "DINGDING_WARNING_PHONE", "DINGDING_WARNING_USER_ID",
          "DINGDING_WARNING_ALL", "DINGDING_WARNING_SECRET",
          "FEISHU_WARNING_URL", "FEISHU_WARNING_USER", "FEISHU_WARNING_ALL",
          "EMAIL_SENDER", "EMAIL_PASSWORD", "EMAIL_RECEIVER", "EMAIL_SMTPSERVER",
          "WECHAT_WARNING_URL", "WECHAT_WARNING_PHONE", "WECHAT_WARNING_ALL",
          "WARNING_INTERVAL", "WARNING_LEVEL", "WARNING_FAILED_COUNT"):
    print("%-26s = %r" % (k, getattr(setting, k)))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
DINGDING_WARNING_URL       = ''
DINGDING_WARNING_PHONE     = ''
DINGDING_WARNING_USER_ID   = ''
DINGDING_WARNING_ALL       = False
DINGDING_WARNING_SECRET    = None
FEISHU_WARNING_URL         = ''
FEISHU_WARNING_USER        = None
FEISHU_WARNING_ALL         = False
EMAIL_SENDER               = ''
EMAIL_PASSWORD             = ''
EMAIL_RECEIVER             = ''
EMAIL_SMTPSERVER           = 'smtp.163.com'
WECHAT_WARNING_URL         = ''
WECHAT_WARNING_PHONE       = ''
WECHAT_WARNING_ALL         = False
WARNING_INTERVAL           = 3600
WARNING_LEVEL              = 'DEBUG'
WARNING_FAILED_COUNT       = 1000

触发点都在框架内部：爬虫开始、爬虫结束、任务成功率低于 0.5、失败数超阈值、任务停滞、入库失败、代理获取失败，统一走 tools.send_msg(msg, level)。

下面这段把「机器人」指到本机的假 webhook，把消息形状、去重、级别门槛一次跑清楚（不发任何外部请求）：

warning_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
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

import redis
from feapder import setting
from feapder.utils import tools

CLIENT = redis.Redis(host="127.0.0.1", port=6390)
for k in list(CLIENT.keys("rate_limit*")):
    CLIENT.delete(k)
HITS = []


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        n = int(self.headers.get("Content-Length", 0))
        HITS.append(self.path + " | " + self.rfile.read(n).decode("utf-8")[:220])
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"ok")

    def log_message(self, *args):
        pass


srv = HTTPServer(("127.0.0.1", 16399), Handler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
URL = "http://127.0.0.1:16399/robot"


def send(msg, level="debug", **kw):
    setting.DINGDING_WARNING_URL = URL
    for k, v in kw.items():
        setattr(setting, k, v)
    tools.send_msg(msg, level=level)
    time.sleep(0.2)


print("== 1. 什么都不配（把 URL 清空）==")
setting.DINGDING_WARNING_URL = ""
tools.send_msg("这条不该外发")
print("   webhook 收到:", len(HITS), "条")

print("== 2. 配好钉钉 URL ==")
send("报警内容A")
print("   收到:", len(HITS), "条")
print("   报文:", HITS[-1])
send("报警内容A")                      # 同内容，落在 WARNING_INTERVAL 里
print("   同内容再发一次，收到:", len(HITS), "条（去重生效则不变）")
send("报警内容B")                      # 不同内容
print("   换内容再发，收到:", len(HITS), "条")
print("   去重键:", [k.decode() for k in CLIENT.keys("rate_limit*")])

print("== 3. WARNING_INTERVAL=0 不去重 ==")
setting.WARNING_INTERVAL = 0
send("报警内容A")
print("   收到:", len(HITS), "条")

print("== 4. WARNING_LEVEL 门槛 ==")
setting.WARNING_INTERVAL = 3600
before = len(HITS)
setting.WARNING_LEVEL = "ERROR"
send("debug 级不该发", level="debug")
send("info 级不该发", level="info")
send("error 级要发", level="error")
print("   本段新增外发:", len(HITS) - before, "条")

print("== 5. @人 / isAtAll / 加签 ==")
setting.WARNING_INTERVAL = 0
send("带手机号与全@", DINGDING_WARNING_PHONE="13800000000", DINGDING_WARNING_ALL=True)
print("   最后一条:", HITS[-1][:200])
setting.DINGDING_WARNING_SECRET = "SEC1234567890"
send("加签后")
print("   加签路径:", HITS[-1].split(" | ")[0][:110])
srv.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
== 1. 什么都不配（把 URL 清空）==
   webhook 收到: 0 条
== 2. 配好钉钉 URL ==
   收到: 1 条
   报文: /robot | {"msgtype": "text", "text": {"content": "feapder报警系统\n报警内容A"}, "at": {"atMobiles": [], "atUserIds": [], "isAtAll": false}}
   同内容再发一次，收到: 1 条（去重生效则不变）
   换内容再发，收到: 2 条
   去重键: ['rate_limit:a8d53e709ef6ebefb73eda95ee3ceed3']
== 3. WARNING_INTERVAL=0 不去重 ==
   收到: 3 条
== 4. WARNING_LEVEL 门槛 ==
   本段新增外发: 1 条
== 5. @人 / isAtAll / 加签 ==
   最后一条: /robot | {"msgtype": "text", "text": {"content": "feapder报警系统\n带手机号与全@"}, "at": {"atMobiles": ["13800000000"], "atUserIds": [], "isAtAll": true}}
   加签路径: /robot&timestamp=1794054322597&sign=sGewFkeA...%3D

四条要点：

  • 不配 URL 就彻底静默：任务失败只写日志，框架一条消息都不发，别以为「没配就没坏」。
  • 去重由 tools.reach_freq_limit 实现：能连上 redis 时写键 rate_limit:<md5(url, phone, 内容)>，TTL 就是 WARNING_INTERVAL；连不上 redis 时退回进程内的字典 freq_limit_record。上面这段没配 redis，所以「去重键」那一行是空的，但同内容确实被吃掉了。本地调试建议临时把 WARNING_INTERVAL 设成 0；配了 redis 时改了配置不生效要先去清那些 rate_limit:* 键。
  • WARNING_LEVEL 默认 DEBUG，意思是「什么都发」；只想收严重告警要显式改成 ERROR。上面第 4 段实测：ERROR 时 debug/info 各一条都不外发，只有 error 那条发了。
  • 钉钉加签是简单字符串拼接 URL&timestamp=...&sign=...，原 URL 没有 query 时拼出来的地址不规范（假服务不在意，真钉钉是否接受请按官方文档复核）。

监控打点是另一套东西（不是开关式的）：feapder.utils.metrics 要 INFLUXDB_HOST/PORT/UDP_PORT/DATABASE/USER/PASSWORD 六个值全部非空才会初始化，否则 emit_* 全部静默 no-op；influxdb 包是模块级硬依赖（精简安装已带上）。实测把参数配全但没有 InfluxDB 服务时，init() 花 6.14 秒超时后仍返回 None，日志里那句拼错的 metrics init falied: ... 可以当作定位线索。打点表名是 task_ + 环境变量 TASK_ID，不设 TASK_ID 时表名是 None，这是 feaplat 管理系统的约定，本地手跑用不上。

工具库 tools

feapder.utils.tools 是框架自带的杂物间，很多细节问题在这里有现成答案。数量口径要说清楚：tools.md 写「共计 129 个」，1.9.3 里按源码 ^def 数是 139 个（含 2 个私有），按 dir() 里可调用对象算是 151 个（多出来的是 RedisDB、EmailSender、md5、partial 这类 import 与装饰器）。挑常用的一批实跑：

tools_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
from feapder.utils import tools as t

print("get_md5('feapder')          :", t.get_md5("feapder"))
print("get_md5('a','b') == get_md5('ab') :", t.get_md5("a", "b") == t.get_md5("ab"))
print("canonicalize_url            :", t.canonicalize_url("HTTPS://X.com/a/../b/./c?b=2&a=1#frag"))
print("key2hump / key2underline    :", t.key2hump("user_name_and_age"), "/", t.key2underline("userNameAndAge"))
print("make_insert_sql             :", t.make_insert_sql("db.tbl", {"id": 1, "name": "张三", "score": 9.5}))
print("make_batch_sql              :", t.make_batch_sql("db.tbl", [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}]))
print("parse_url_params            :", t.parse_url_params("https://x.com/a?b=1&c=%E4%B8%AD"))
print("get_param                   :", t.get_param("https://x.com/a?b=1&c=2", "c"))
print("del_html_tag                :", t.del_html_tag("<p class='a'>Hello<b>World</b>&nbsp;中文</p>"))
print("del_html_js_css             :", t.del_html_js_css("<script>var a=1</script><style>p{}</style><div>正文</div>"))
print("cookies2str / get_cookies_from_str :", t.cookies2str({"sid": "abc", "u": "zhang"}), "/",
      t.get_cookies_from_str("sid=abc; u=zhang"))
print("get_json 正常 / 非 json       :", t.get_json('{"a": 1}'), "/", t.get_json("不是json"))
print("dumps_json                  :", t.dumps_json({"中文": "值", "n": 1}, indent=None))
print("cut_string                  :", t.cut_string("abcdefghij", 4))
print("format_seconds              :", t.format_seconds(3726.5))
print("date_to_timestamp / 反向      :", t.date_to_timestamp("2026-01-01 00:00:00"), "/",
      t.timestamp_to_date(1767196800))
print("format_time                 :", t.format_time("2021-01-01 00:00:00", "%Y%m%d"))
print("unquote_url / quote_url     :", t.unquote_url("%E7%88%AC%E8%99%AB"), "/", t.quote_url("爬虫"))
print("is_have_chinese / get_chinese_word :", t.is_have_chinese("abc中文"), "/",
      t.get_chinese_word("abc中文def你好"))
print("get_full_url / get_domain   :", t.get_full_url("https://a.com/x/y", "/z?i=1"), "/",
      t.get_domain("https://a.com:8080/x?q=1"))
print("ip_to_num / get_localhost_ip:", t.ip_to_num("192.168.1.1"), "/", t.get_localhost_ip())
print("get_base64 / get_sha1 / get_random_string :", t.get_base64("feapder"), "/",
      t.get_sha1("feapder")[:12], "/", t.get_random_string(8))
print("dumps_obj → loads_obj 往返   :", t.loads_obj(t.dumps_obj({"x": [1, 2], "s": "中"})))
print("is_valid_url 三条           :", [t.is_valid_url(u)
      for u in ("https://a.com", "javascript:void(0)", "")])
print("import_cls                  :", t.import_cls("feapder.network.response.Response"))
print("get_method 存在 / 不存在      :", bool(t.get_method(t, "get_md5")), "/", t.get_method(t, "不存在"))
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
get_md5('feapder')          : 3506912ade3ceceed54170c84663b92c
get_md5('a','b') == get_md5('ab') : True
canonicalize_url            : https://x.com/b/c?a=1&b=2
key2hump / key2underline    : UserNameAndAge / user_name_and_age
make_insert_sql             : insert into `db.tbl` (`id`, `name`, `score`) values (1, '张三', 9.5)
make_batch_sql              : ('insert ignore into `db.tbl` (`name`, `id`) values (%s, %s)', [['a', 1], ['b', 2]])
parse_url_params            : ('https://x.com/a', {'b': '1', 'c': '中'})
get_param                   : 2
del_html_tag                : HelloWorld中文
del_html_js_css             : <div>正文</div>
cookies2str / get_cookies_from_str : sid=abc; u=zhang;  / {'sid': 'abc', 'u': 'zhang'}
get_json 正常 / 非 json       : {'a': 1} / {}
dumps_json                  : {"中文": "值", "n": 1}
cut_string                  : ['abcd', 'efgh', 'ij']
format_seconds              : 1小时2分7秒
date_to_timestamp / 反向      : 1767196800 / 2026-01-01 00:00:00
format_time                 : 20210101
unquote_url / quote_url     : 爬虫 / %E7%88%AC%E8%99%AB
is_have_chinese / get_chinese_word : True / ['中文', '你好']
get_full_url / get_domain   : https://a.com/z?i=1 / a.com:8080
ip_to_num / get_localhost_ip: 3232235777 / 192.168.1.39
get_base64 / get_sha1 / get_random_string : ZmVhcGRlcg== / b78b6f08cc7b / O2vsizpR
dumps_obj → loads_obj 往返   : {'x': [1, 2], 's': '中'}
is_valid_url 三条           : [True, False, False]
import_cls                  : <class 'feapder.network.response.Response'>
get_method 存在 / 不存在      : True / None

几个必须记住的口径：

  • cut_string 是「按长度切成列表」，不是截断字符串；format_time 的第一个位置参数不是时间戳（传时间戳返回 None）；get_param、get_json 都没有 default= 形参；没有 cookie_str_to_dict（叫 get_cookies_from_str）、没有 parse_params（叫 parse_url_params）。
  • dumps_obj/loads_obj 是 pickle 包装，request 携带字典等复杂对象进 redis 走的就是它，所以那些值必须可 pickle。
  • is_valid_proxy('1.2.3.4:8080') 会真的联网探测，超时就是 check proxy failed: timed out，不要在生产脚本里随手调用。
  • get_json 解不出来时返回 {}，同时打两条 ERROR 日志（不是抛异常），静默错数据往往从这里来。

与 Scrapy 怎么选

同一个需求（抓 ssr1 列表 + 详情 + 落库），两套框架的差别主要在这些地方：

维度 feapder 1.9.3 Scrapy 2.19
并发模型 线程池（SPIDER_THREAD_COUNT） asyncio + Twisted
任务存储 Redis zset（score 兼作优先级与防丢时间） Scrapy Scheduler；分布式要 scrapy-redis
断点续爬 默认能力，redis_key 复用即可 需 scrapy-redis，且 start_requests 会重新下发
数据落库 Item + ITEM_PIPELINES（mysql/mongo/csv/console） Item + Pipeline（要自己写 SQL/驱动）
去重 内置 Dedup 四种过滤器，可选永久/临时/内存/轻量 默认指纹集合（内存），跨进程需自定义
浏览器渲染 render=True 一行开关（selenium/playwright） 需 scrapy-playwright 等中间件
报警 内置钉钉/飞书/企微/邮件，配 URL 即发 需自己接 Sentry/日志告警
周期批次 BatchSpider 自带批次表与 batch_date 无对应物
配置 103 个键，setting.py/__custom_setting__/环境变量三处 settings.py + 中间件/扩展装配
学习曲线 接口少、封装深，出问题要读它的源码 接口多、显式，生态和文档更成熟

选型建议：

  • 数据量不大、要快速出活：requests + 解析（见《爬虫-基本入门》《爬虫-Requests》）或 AirSpider，别上分布式。
  • 需要断点续爬、报警、渲染、批次采集，且团队接受「约定优于配置」：feapder 的封装能省掉 scrapy-redis + 报警 + 渲染中间件这一整套胶水。
  • 需要精细控制下载器、中间件链、复用生态（爬虫平台、LinkExtractor、ImagesPipeline 等）：Scrapy 更合适，见《爬虫-Scrapy》。
  • 两者都不是「异步写法的爬虫」，feapder.start() 在本机 Python 3.14 上还会 join 同进程的非守护线程——要嵌进 Web 服务时，异步框架（FastAPI/Tornado）与爬虫进程分开部署。

常见报错速查

现象 / 报错 原因 处理
ModuleNotFoundError: No module named 'distutils'（import feapder 时） 依赖锁 redis<4，老 redis 客户端 import distutils，Python 3.12+ 已删 在该环境里 pip install setuptools
pip install feapder 后原环境的 redis 版本变低 redis<4.0.0 约束，pip 静默降级 给 feapder 单独建 venv
RuntimeError: terminal_layout can only run on Terminal feapder create -s/-i 是方向键菜单 在真终端里跑，或直接调 CreateSpider().create(name, type)
ConnectionError: Error 10061 connecting to ... Redis 没起/端口密码错 起服务、核对 REDISDB_*
redis.exceptions.ConnectionError: 未设置 redis 连接信息 用到了 redis 的功能但既没 setting.py 也没环境变量 配 REDISDB_IP_PORTS（构造时就抛，不是使用时）
AttributeError: 'MysqlDB' object has no attribute 'connect_pool' + ip: None MySQL 连接信息为空（含 update_task_batch 误用在 redis 任务表上） 配 MYSQL_* 环境变量或删掉该 yield
ModuleNotFoundError: No module named 'pymongo' 精简版没有 mongo 管道依赖 pip install "feapder[all]"
Exception: 需要安装feapder完整版 filter_type=2（MemoryFilter）要 bitarray 装完整版或改 filter_type=3/4
ModuleNotFoundError: No module named 'playwright' 渲染包会同时 import 两种驱动 渲染就按 feapder[render] 装齐
TypeError: 'XxxParser' object is not callable add_parser 传了实例 传类，且集成模式下别再用「继承 Spider 写 start_requests」
TypeError: 'dict' object is not callable 把 response.json 当方法调用 它是 property，写 response.json
TypeError: eval() arg 1 must be a string...（用户池） GoldUserPool 没重写 login()，池里没人 实现 login()；否则进程会无限刷并卡住
某条 URL 一条请求都不发 parser_name 写错被静默丢弃；或去重库判重（尤其全局 dedup:bloomfilter:bloomfilter0） 检查类名；delete_keys/换 name/filter_repeat=False
重启后卡在「等待任务」 任务被取走时 score 改成了 now+600，10 分钟内不可见 delete_keys="*z_requests" 或等 REQUEST_LOST_TIMEOUT 过期
渲染拿到的页面没数据 render_time 默认 0，JS 还没填完 给 render_time=3 以上，或改走接口
代理请求全失败但看不出错 API 正文用 \n 分隔，被当成一个代理 换成 \r\n；返回 JSON/HTML 时框架只按「含 { 或空」判失败
response.status_code 在渲染下总是 200 渲染下载器写死了 200、elapsed=666 用正文特征或 validate 判断，别看状态码
改了 setting.LOG_LEVEL 没反应 日志器在第一条日志那一刻就初始化完了 在爬虫类里用 __custom_setting__ 配好，或程序最开始改
第二次跑同样数据一条都没入库 item/request 去重库记住了上次结果 看 dedup:* 键；必要时换 name 或清库
跑完不报错，MySQL 里却没新增行 批量入库默认走 insert ignore，重复行被静默跳过（affected rows 0），INFO 级别什么都不打 想更新就换 UpdateItem；排查时把 LOG_LEVEL 调成 DEBUG，会看到「没有插入数据，可能全部重复」
update_task_batch 之后 state 一直是 2，日志刷 MysqlPipeline 更新数据失败 它只往任务表插 {id, state} 两个字段，表里还有别的 NOT NULL 列就当场失败 非空列改可空（官方示例的 param 就可空），或改用 yield lambda: self.update_task_state(task_id)
硬杀 worker 重启后，任务既不补发也不重置 那几条在 redis 里 score=now+600，仍被 zcard 计入总数，master 只打「有积压任务，暂时不派发新任务」 等 600 秒；或 delete_keys="*z_requests" 配合手工 update 任务表 set state=0 where state=2；或调小 REQUEST_LOST_TIMEOUT
批次表里没有 batch_name 列 batch_name 只是程序里的标签，不落库 批次进度看 done_count / is_done；数据要带批次时间得自己 item.batch_date = self.batch_date

参考