爬虫监控告警实战：企业微信+钉钉机器人搭建7x24小时守护

1. 爬虫为什么必须有一套监控告警体系

先讲个真实场景。你写了一个爬虫，每天凌晨两点定时去抓竞品价格数据，平时跑得好好的，突然某天对方网站改版了页面结构，你的解析规则全部失效，爬虫开始疯狂报错或者更糟——静默抓回一堆空数据。如果这个爬虫没有监控，你可能要到第二天早上打开数据库，看到连续五六个小时的数据全是空的，才意识到出事了。更麻烦的是，这种问题往往不是一次性的，网站改版、IP被限制、验证码升级、接口限流，每一个都能让你的爬虫悄悄"病死"。

我这些年维护过的爬虫项目，几乎每个都经历过这种"半夜翻车"的时刻。爬虫本身不难写，难的是让它在无人值守的情况下持续稳定运行。所以我才说，爬虫监控告警不是可选项，是必选项。这篇文章要聊的，就是怎么用企业微信或钉钉的机器人，搭配一套轻量级的监控逻辑，给爬虫装上 7×24 小时的"生命体征监测仪"。

这套系统适合谁？适合手里有爬虫任务在跑的个人开发者、运维工程师、数据团队。不管你的爬虫是跑在云服务器上，还是跑在公司内网机器上，只要你能执行 Python 脚本、能访问外网 API，这套方案就能直接落地。它解决的核心问题有三个：第一时间知道爬虫挂了、第一时间知道数据异常了、不用半夜爬起来看日志。

2. 爬虫监控的整体架构与关键指标设计

2.1 监控数据从哪来：爬虫的四个关键信号

想监控爬虫，先得搞清楚监控什么。我见过不少人一上来就盯着"进程还在不在"，其实进程活着不代表爬虫正常，进程死了也不一定需要立刻报警——得看场景。

我建议把爬虫的监控指标分成四层：

第一层是进程级。进程是否存活、有没有被系统 OOM Killer 干掉、主循环是否卡死。这一层最简单，但只覆盖"爬虫跑没跑"的问题。

第二层是任务级。单次任务是否成功完成、耗时是否异常、有没有重试、最终状态是成功还是失败。这一层解决的是"这一轮抓取任务是否正常结束"。

第三层是数据级。这是最容易忽略、却最重要的一层。抓回来的数据量是否在正常区间？新增记录数是零还是突然暴涨？去重率是不是异常？字段完整性怎么样？有时候爬虫明明"成功"了，但因为页面结构变化，解析出来的全是空字段，任务状态显示成功，数据却是一堆垃圾。没有数据级的监控，这个坑你根本发现不了。

第四层是资源级。CPU、内存、磁盘、带宽。爬虫是资源消耗大户，尤其是跑在共享服务器上的爬虫，容易被其他任务挤死，或者因为日志膨胀把磁盘写满。

我自己的实践中，这四层指标会映射成几条具体的告警规则，每一条都能对应到一个可执行的检查动作。比如进程级用 supervisor 或 systemd 来守护和检查，任务级在爬虫入口和出口埋点上报，数据级对比历史均值做波动检测，资源级直接采集系统指标。

2.2 告警规则与阈值怎么定

阈值定得太松，等于没装监控；定得太紧，每天告警刷屏，几天后你就开始无视所有告警。这里分享几个我踩过坑之后的经验。

第一条规则是不要用固定绝对值，要用滑动基线。比如"单次任务抓取条数低于 100 就告警"，这个规则看起来很合理，但如果你的爬虫抓的是一个流量波动很大的网站，周末数据量本来就低，这个阈值就会在周末疯狂误报。更好的做法是取最近 7 天同一时段的均值，低于均值的 30% 才告警。实现上也不难，从数据库里查一下最近七天的记录做个 AVG 就行。

第二条规则是连续失败才告警。单次失败可能是网络抖动、目标网站临时抽风，不值得把人从被窝里叫起来。我常用的策略是"连续 3 次失败"或者"10 分钟内失败超过 2 次"才触发告警。用代码实现就是在内存里维护一个失败计数器，重置条件有两个：成功一次清零，或者超过一段时间自动衰减。

第三条规则是区分告警级别。我把告警分成三级：WARN 级只发消息不打扰，比如单次任务失败但重试成功；ERROR 级需要尽快处理，比如连续失败、数据量为零；CRITICAL 级必须立刻处理，比如进程挂了、磁盘快满了。不同级别可以走不同的推送策略，低级别合并发送，高级别单独秒推。

2.3 告警降噪：避免凌晨三点被吵醒

告警降噪这个词听起来高大上，其实核心就一句话：让每条告警都有足够的信息量，并且避免重复轰炸。

最常见的降噪手段是去重聚合。同一个任务在 10 分钟内连续告警 5 次，正常人只需要看到一条消息"任务 X 连续失败 5 次"，附带最近一次的错误信息就够了。实现方式是在告警模块里维护一个字典，key 是任务名加错误类型，value 是首次触发时间和累计次数。推送时如果发现同样的 key 在聚合窗口内已经推过，就不再推送，只更新计数器。等到窗口结束，再补推一条聚合结果。

另一个手段是静默窗口。有些告警你知道了也没法立刻处理，比如目标网站的反爬策略变化，你得等到上班才能分析。这种情况下，可以给告警规则配置一个"工作时间外只记录不推送"的选项，或者把夜间告警统一合并成早上的"夜间告警汇总"。

还有一个容易被忽略的降噪点：恢复通知。故障恢复之后，必须给运维人员发一条"已恢复"的消息，否则大家不知道问题解决了，可能白忙一场。恢复通知也要做去重——只有之前推送过故障告警的任务，恢复时才推送。

3. 告警推送通道的搭建实操

3.1 企业微信机器人：从创建群聊到拿到 Webhook

企业微信机器人是目前我觉得最省事的告警通道。它不需要单独安装客户端，也不需要申请什么应用权限，只要你能建一个企业微信群，就能创建一个机器人，拿到一个 Webhook 地址。

具体步骤很简单：先在企业微信里创建一个群（可以只有你自己一个人），然后在群设置里找到"群机器人"，点击"添加机器人"，给它起个名字，系统会生成一个 Webhook 地址，形如https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx。

拿到 Webhook 之后，推送消息只需要一个 HTTP POST 请求。企业微信机器人支持文本（text）、Markdown（markdown）、图片（image）、图文（news）、文件（file）五种消息类型。对告警场景来说，最常用的是 text 和 markdown 两种。

有一个很实用的细节：markdown 消息的正文会被包裹在一个markdown字段里，而且它的语法支持有限，不支持 HTML。我一般会把告警信息格式化成文本加简单 Markdown 的混合体，比如用**加粗**标注任务名、用>引用错误信息，这样的阅读体验比纯文本好很多。另外，企业微信机器人的消息内容默认不会自动换行，需要显式在内容里加\n换行，这个坑我第一次踩的时候折腾了十分钟。

推送的时候，只需要用 requests 发 POST 请求：

import requests import json def send_wecom(webhook: str, content: str, msg_type: str = "text"): if msg_type == "text": payload = { "msgtype": "text", "text": {"content": content} } else: payload = { "msgtype": "markdown", "markdown": {"content": content} } resp = requests.post(webhook, json=payload, timeout=10) result = resp.json() if result.get("errcode") != 0: raise RuntimeError(f"企业微信推送失败: {result.get('errmsg')}") return result

这里有几个细节。第一是超时时间必须设置，不然企业微信接口万一 hang 住，你的告警线程也跟着卡死。第二是返回值里的errcode，只有等于 0 才是成功，其他值都有具体含义，比如93000表示 webhook 不存在或已被删除，93001表示机器人已被移出群聊。把这些错误码记录下来，排查问题的时候会省很多时间。

3.2 钉钉机器人：安全设置与加签方式

钉钉机器人的整体流程和企业微信类似，也是在群里添加自定义机器人，然后拿到 Webhook 地址。不过钉钉多了一个"安全设置"环节，有三种方式：自定义关键词、加签、IP 白名单。我强烈建议用加签方式，因为它不需要额外维护 IP 列表，安全性也足够。

加签的原理是：你设置一个密钥（Secret），推送时用当前时间戳加上密钥做 HMAC-SHA256 签名，然后把签名值拼到 Webhook URL 后面。钉钉服务器会校验这个签名，校验通过才会接受消息。签名算法固定是：

import time import hmac import hashlib import base64 import urllib.parse def dingtalk_sign(secret: str) -> str: timestamp = str(round(time.time() * 1000)) secret_enc = secret.encode("utf-8") string_to_sign = f"{timestamp}\n{secret}".encode("utf-8") hmac_code = hmac.new(secret_enc, string_to_sign, digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return timestamp, sign

推送时，把 Webhook 地址、&timestamp=xxx&sign=xxx拼在一起再 POST 数据。钉钉的 payload 结构和企业微信略有不同，文本消息用 text/markdown 包一层：

def send_dingtalk(webhook: str, secret: str, content: str, msg_type: str = "text"): timestamp, sign = dingtalk_sign(secret) url = f"{webhook}&timestamp={timestamp}&sign={sign}" if msg_type == "text": payload = { "msgtype": "text", "text": {"content": content} } else: payload = { "msgtype": "markdown", "markdown": {"title": "爬虫告警", "text": content} } resp = requests.post(url, json=payload, timeout=10) result = resp.json() if result.get("errcode") != 0: raise RuntimeError(f"钉钉推送失败: {result.get('errmsg')}") return result

有个容易踩的坑：钉钉对消息体的限制比企业微信严格，文本消息最大支持 5000 字节，Markdown 消息有限制，图片消息对 URL 有域名白名单要求。我之前就遇到过因为告警信息里塞了太长的错误堆栈，导致推送一直失败，后来才知道是被长度限制卡住了。处理办法是在推送前对消息做截断，保留前 1500 个字符，超出的部分用...（完整日志见 server:/path/to/log）代替。

还有一个细节：钉钉机器人每秒只能发送 20 条消息，虽然正常情况根本达不到这个量级，但如果你从"批量告警"变成"批量轰炸"，就会被限流。所以告警推送模块里最好加一个简单的并发控制，用threading.Semaphore或者直接串行推送即可。

3.3 Python 封装统一的告警推送模块

企业微信和钉钉各自封装一套代码是可以的，但如果你以后想切换或者同时用两个渠道，维护成本就上来了。我更推荐的做法是做一个统一的告警客户端接口，内部再分发到具体渠道。这样业务代码里只需要调用alert_client.send("task_failed", task_name, error_msg)，完全不用关心底层走的是企业微信还是钉钉。

class AlertClient: def __init__(self, wecom_webhook=None, dingtalk_webhook=None, dingtalk_secret=None): self.wecom_webhook = wecom_webhook self.dingtalk_webhook = dingtalk_webhook self.dingtalk_secret = dingtalk_secret self._recent_alerts = {} # (task_name, alert_type) -> last_sent_time def send(self, task_name, alert_type, message, level="ERROR"): key = (task_name, alert_type) now = time.time() last_time = self._recent_alerts.get(key, 0) if now - last_time < 300: # 五分钟内同一类型不重复推送 return False content = self._format_message(task_name, alert_type, message, level) errors = [] if self.wecom_webhook: try: send_wecom(self.wecom_webhook, content, msg_type="markdown") except Exception as e: errors.append(f"企业微信: {e}") if self.dingtalk_webhook: try: send_dingtalk(self.dingtalk_webhook, self.dingtalk_secret, content, msg_type="markdown") except Exception as e: errors.append(f"钉钉: {e}") if errors: # 两个渠道都失败时，写入本地日志，避免告警丢失 log_error(f"告警推送失败: {errors}") return False self._recent_alerts[key] = now return True

这个模块虽然简单，但把去重、多渠道分发、失败兜底这几件事都做了。尤其是"两个渠道都失败"的情况，我建议一定要在本地留一份日志，否则告警本身就丢了，你连"告警发送失败"这件事都不知道，那才是真正的灾难。

4. 存储、调度与完整链路串起来

4.1 用 SQLAlchemy 记录爬虫运行日志与告警历史

告警消息推出去就完了吗？不是的。我强烈建议把每次爬虫任务的运行状态、每次告警的触发记录都存到数据库里。有两个目的：一是做数据级的基线计算（前面提到的滑动均值），二是事后回溯问题时，你能查到"这个任务在什么时间点开始失败、失败了多少次、每次的错误信息是什么"。

存储方案我用的是 SQLAlchemy，它作为一个 ORM 层，可以无缝切换 SQLite、MySQL、PostgreSQL。对于爬虫监控这种场景，量级通常不大，SQLite 单文件就能跑，但如果你已经有 MySQL 了，直接连上也不麻烦。

先定义两个核心模型：

from datetime import datetime from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, Float from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() class CrawlTaskLog(Base): __tablename__ = "crawl_task_log" id = Column(Integer, primary_key=True, autoincrement=True) task_name = Column(String(128), nullable=False, index=True) status = Column(String(16), nullable=False) # success / failed / timeout started_at = Column(DateTime, nullable=False) finished_at = Column(DateTime) item_count = Column(Integer, default=0) # 抓取条数 error_msg = Column(Text) # 失败时的错误信息 duration_sec = Column(Float) # 任务耗时 class AlertRecord(Base): __tablename__ = "alert_record" id = Column(Integer, primary_key=True, autoincrement=True) task_name = Column(String(128), nullable=False) alert_type = Column(String(32), nullable=False) # task_failed / data_empty / process_down level = Column(String(8), nullable=False) # WARN / ERROR / CRITICAL content = Column(Text) created_at = Column(DateTime, nullable=False, default=datetime.now)

写入逻辑挂在爬虫任务的 finally 块里，无论成功失败都记录一条。有个细节要注意：如果爬虫本身崩得很惨，连 finally 都没执行到怎么办？这就需要在爬虫外层再用一个守护进程来兜底，后面讲调度的时候会提到。

基线计算就是基于CrawlTaskLog表的历史数据做的。比如判断"这次抓取条数是否异常偏低"，就查最近 7 天同任务名的成功记录，算平均值和标准差，如果本次值低于均值减两倍标准差，就认为异常。这个逻辑用 SQLAlchemy 写起来很清晰：

from sqlalchemy import select, func from datetime import timedelta def check_item_count_anomaly(session, task_name, item_count, days=7): since = datetime.now() - timedelta(days=days) result = session.execute( select(func.avg(CrawlTaskLog.item_count), func.stddev(CrawlTaskLog.item_count)) .where(CrawlTaskLog.task_name == task_name) .where(CrawlTaskLog.status == "success") .where(CrawlTaskLog.started_at >= since) ).one() avg, std = result if avg is None or item_count < avg - 2 * std: return True return False

注意stddev函数在 SQLite 里是原生支持的，在 MySQL 里叫STDDEV，PostgreSQL 里叫STDDEV，SQLAlchemy 会帮你翻译，不用操心。但如果样本太少（比如历史上就两三次成功记录），算出来的标准差可能失真，所以我在代码里会加一个if count < 5: return False的保护逻辑，样本太少时不告警，避免误报。

4.2 定时调度巡检：crontab 与 DolphinScheduler 两种方案

告警系统本身需要有一个调度机制来驱动。这里分两个层面：爬虫任务的定时执行，以及监控巡检的定时执行。

最简单的方案是 crontab。比如你的爬虫每天凌晨 2 点跑一次，就在 crontab 里写：

0 2 * * * cd /opt/spider && /usr/bin/python3 run_task.py >> logs/task.log 2>&1

然后在run_task.py里埋好各种状态上报逻辑。这种方式对单个或少量爬虫足够用，但 crontab 有个缺点：没有主动的重试机制，爬虫进程因为环境问题挂掉之后，下一次执行还得等下一个周期。

进阶一点的方案是用 DolphinScheduler 这类工作流调度平台。它自带失败重试、超时告警、依赖管理，而且可以直接配置"任务失败后发起告警"的钩子。DolphinScheduler 里创建一个 Shell 节点来跑爬虫命令，然后在"失败策略"里选"告警"，在告警组里配置企业微信或钉钉的 Webhook，调度平台本身就把"任务失败→发告警"这条链路打通了。如果你的爬虫任务比较多、执行流复杂，我建议直接用 DolphinScheduler，它能省掉你自己写调度和重试逻辑的时间。

不过，调度平台只管"任务是否按计划执行"，它不管"任务执行了但数据异常"。所以无论用哪种调度方案，我都建议另外跑一个独立巡检脚本，每 10 分钟检查一次CrawlTaskLog表：如果某个任务本该在特定时间执行完，却查不到对应记录，说明任务根本没启动，这时立即告警。这个巡检脚本本身也放在 crontab 或 systemd timer 里，形成一主一备的双重保障。

4.3 消息格式设计：让告警一眼看懂

告警消息的格式，直接决定了收到消息的人能不能在 10 秒内判断出问题的严重性。我见过最糟糕的消息是只有一行": 时间"，连哪个任务、什么错误都不知道。

我自己的格式化模板是固定的，经过多次迭代后，现在的版本长这样：

**[CRITICAL] 爬虫任务连续失败** - 任务名称：product_price_crawler - 失败次数：连续 3 次 (时间窗口: 14:00-14:30) - 最新错误：AttributeError: 'NoneType' object has no attribute 'find_all' - 最近成功：2025-01-05 14:00:00 - 影响范围：products 表昨日价格字段缺失 - 处理建议：检查目标站页面结构是否变更

这个格式有几个设计要点：第一行必须有级别和任务名，让人一眼定位；错误信息必须是最新的，不是最早的那条；"最近成功"时间让运维判断问题的持续时间；"影响范围"让运维判断要不要立即处理还是可以等上班再说。

实现上，任务名和错误信息由爬虫传进来，影响范围和处理建议可以由一个简单的规则映射生成。比如提前维护一个任务清单：

TASK_META = { "product_price_crawler": { "name": "商品价格抓取", "impact": "products 表价格字段缺失，影响比价功能", "suggestion": "检查目标站结构 + 确认解析规则", }, }

有了这个映射，告警内容就从"技术日志"变成了"业务可读的信息"。这一点在团队协作时尤其重要——不是每个收到告警的人都是写这个爬虫的人，但格式化的信息能让任何接手的人快速理解发生了什么。

5. 常见问题与排查实录

5.1 机器人消息发不出去

告警推送模块写完上线，第一时间要做的不是等告警，而是主动测试。我建议在代码里留一个--test参数，跑一次真实推送。实测中我遇到过几类问题，现在整理成速查表：

这里单独提一下时间戳的问题。加签模式下，钉钉会校验timestamp与服务器时间差是否在 1 小时内。如果服务器时间漂移严重，签名判断就会失败。我遇到过一台 NTP 服务挂掉的服务器，时间慢了两个小时，所有钉钉告警全部发送失败，故障恢复后那台机器的时间校准，问题才消失。所以这类系统依赖时钟的，一定要把 NTP 同步纳入日常巡检。

5.2 告警刷屏与重复告警

告警刷屏的根源往往是去重逻辑没做好。最容易出现的情况是：爬虫任务失败后，错误被异常捕获，然后在一个大循环里重复上报。比如你的爬虫是分页抓取，每一页失败都会抛一个异常，如果你在每页的异常处理里都调用一次alert_client.send()，那一次任务失败就能刷出几十条告警。

解决思路有两个层面。第一层是在爬虫层面做错误聚合：一整个任务不管失败多少次，只发一次告警，错误信息里带上失败页数和最早/最新的错误样例。第二层是在告警客户端层面做时间窗口去重，也就是前面代码里self._recent_alerts那个字典。

另外还有一个告警升级的逻辑可以加：第一次失败发 WARN，第 N 分钟还没恢复发 ERROR，超过 1 小时发 CRITICAL。这种"升级式告警"比一上来就 CRITICAL 体验好得多，因为很多问题确实能在几分钟内自动恢复，没必要搞得全组紧张。

5.3 爬虫长期运行的稳定性坑点

监控系统做得再好，也架不住爬虫本身埋的雷。这里说几个我在长期运维中反复遇到的坑。

第一个坑是日志文件无限膨胀。爬虫如果用了print()调试，输出重定向到日志文件之后，文件会越来越大，最后磁盘满，爬虫写不了数据直接闪崩。解决办法是日志轮转，用 Python 的logging.handlers.RotatingFileHandler设置单文件大小和备份数量，比如单文件 50MB、保留 5 个备份。

第二个坑是数据库连接泄漏。如果爬虫在循环里反复创建 SQLAlchemy session 而不关闭，连接池最终会被耗尽，导致"数据库连接数超限"的假性故障。这个问题的现象很迷惑人，因为数据库本身没挂，但新请求全部排队。解决办法是with sessionmaker() as session:这种上下文管理器用法，确保 session 用后必关。

第三个坑是内存缓慢增长。爬虫在长时间运行中，如果某些大对象没有及时释放，会逐渐吃光内存。这种问题监控里很难自动发现，因为进程没死，只是越来越慢。我的处理办法有两个方向：一是用tracemalloc做定期快照，对比内存分配热点；二是更务实地给爬虫设置"跑到一定时间自动重启"的策略，比如单进程最多运行 6 小时，到点主动退出，由调度平台重新拉起。对于很多爬虫场景，定期重启是性价比极高的稳定性方案。

第四个坑是反爬策略变化导致的"假死"。目标网站可能突然让你所有请求都返回 200 但内容是验证码页面，此时爬虫不会报错，反而会抓回一堆解析不了的数据。这种问题只有数据级监控能发现——对比历史抓取条数、字段非空率，一旦指标异常下跌就立刻告警。这也是我前面反复强调数据级监控的原因，它往往是最后一道、也是最关键的一道防线。

6. 最后的几点体会

这套系统从最初的"只在任务失败时发一条企业微信消息"，演进到现在的"四层指标 + 多级告警 + 降噪聚合 + 数据基线校验"，中间改了很多轮。我自己最大的体会是：监控告警系统的价值不在于技术多复杂，而在于"敢不敢在半夜放心睡觉"。当你把任务状态、数据异常、资源水位、推送通道全都打通并验证过之后，那种踏实感是写多少行爬虫代码都换不来的。

如果你想快速起步，建议不要一口气追求完美。先做最小闭环：一个爬虫任务、一次状态上报、一个企业微信机器人、一条失败告警。跑通之后，再继续加数据量校验、加告警降噪、加调度平台。我个人的经验是，80% 的收益来自 20% 的投入，把最基本的"失败告警"和"数据异常告警"做好，就已经超过了大多数裸奔的爬虫项目。

最后再分享一个小技巧：给告警机器人建一个专用的测试群，和正式业务群分开。每次改动告警模板或推送逻辑，先往测试群发一条，确认格式和内容都没问题，再切到正式环境。我因为图省事，直接在正式群里调告警消息格式，结果格式错误的消息把全组人都震了一遍，之后学乖了，测试群这个习惯一直保留到现在。