当前位置：首页 > 文章列表 > 文章 > python教程 > Python爬虫日志监控与告警方法

Python爬虫日志监控与告警方法

2026-02-04 21:09:42 0浏览收藏

积累知识，胜过积蓄金银！毕竟在文章开发的过程中，会遇到各种各样的问题，往往都是一些细节知识点还没有掌握好而导致的，因此基础知识点的积累是很重要的。下面本文《Python爬虫日志监控与异常告警技巧》，就带大家讲解一下知识点，若是你对本文感兴趣，或者是想搞懂其中某个知识点，就请你继续往下看吧~

requests.get()成功但页面内容为空，需先查日志确认请求是否被服务器正确接收并返回预期内容，重点检查URL编码、headers一致性、timeout设置及异常捕获。

Python爬虫日志与监控方案_异常告警技巧【技巧】

为什么 requests.get() 成功但页面内容为空？先查日志再定位

爬虫看似运行成功，response.status_code 是 200，但 response.text 却是空字符串、跳转页或反爬提示页——这种“静默失败”最消耗排查时间。关键不是立刻改代码，而是确认请求是否真被目标服务器接收并返回了预期内容。

必须在发起请求前记录完整上下文：

url（注意是否含中文或特殊字符，需检查是否被自动编码）
headers 中的 User-Agent、Cookie、Referer 是否与浏览器一致
timeout 值是否过小导致连接中断却未抛异常（requests 默认无超时，建议显式设为 timeout=(3, 7)）
捕获 requests.exceptions.RequestException 及其子类，而非只抓 Exception

import logging
import requests
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(name)
try:
resp = requests.get(url, headers=headers, timeout=(3, 7))
logger.info(f"GET {url} → {resp.status_code}, len={len(resp.content)}")
if not resp.content.strip():
logger.warning(f"Empty response from {url}")
except requests.exceptions.Timeout:
logger.error(f"Timeout on {url}")
except requests.exceptions.ConnectionError:
logger.error(f"Connection failed for {url}")

用 logging.Filter 精准过滤爬虫日志，避免告警疲劳

高频爬取时，每秒数条 INFO 日志会淹没真正的问题；而把所有 ERROR 都发邮件/钉钉，又容易因网络抖动触发误报。靠 logging.Filter 实现「只对特定错误模式告警」才是可持续方案。

例如：只对连续 3 次 503 Service Unavailable 或单次 429 Too Many Requests 触发告警，其余归档到文件即可。

继承 logging.Filter，重写 filter() 方法，用实例变量缓存最近 N 次响应码
避免在 filter 中做耗时操作（如写数据库、发 HTTP 请求），否则阻塞主线程
将告警逻辑移出 filter，改为定时扫描日志文件或用 QueueHandler 异步处理

class HttpStatusFilter(logging.Filter):
    def __init__(self, name='', window_size=3):
        super().__init__(name)
        self.status_history = []
        self.window_size = window_size
def filter(self, record):
    if hasattr(record, 'status_code'):
        self.status_history.append(record.status_code)
        self.status_history = self.status_history[-self.window_size:]
        # 触发告警条件（不在此处发送，仅标记）
        if record.status_code == 429 or self.status_history.count(503) >= 3:
            record.needs_alert = True
    return True

监控 requests 耗时突增？用 timeit + 分位数比平均值更可靠

用 time.time() 差值统计单次请求耗时，再算平均值来判断“变慢”，会受偶发长尾请求干扰。比如 99% 的请求在 800ms 内完成，但某次 DNS 解析失败卡了 15s，平均值就被拉高，误判为服务退化。

真实可用的耗时监控应基于分位数（如 p95、p99）和突变检测：

用 time.perf_counter() 替代 time.time()，精度更高且不受系统时间调整影响
每分钟聚合一次耗时数据，计算 p95 并与前一小时同时间段的 p95 对比，浮动超 200% 才触发告警
记录 connect 和 read 阶段耗时（通过 requests.adapters.HTTPAdapter 的 max_retries 和自定义 Response 钩子）

import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class TimingHTTPAdapter(HTTPAdapter):
def send(self, request, kwargs):
start = time.perf_counter()
r = super().send(request, kwargs)
end = time.perf_counter()
r.elapsed_total = end - start
return r
session = requests.Session()
adapter = TimingHTTPAdapter()
session.mount("http://", adapter)
session.mount("https://", adapter)