纯Python实现秒级IP访问突增检测：通过tail-F流式读取Nginx日志，正则提取IP并过滤内网与健康检查；用deque维护60秒滑动窗口计数，动态阈值判断突增（初始50次/60秒，超均值300%即告警）。
核心思路：流式解析 + 滑动时间窗口 + 阈值动态判断
不依赖 Elasticsearch 或 Kafka 等中间件，纯 Python 脚本即可实现秒级响应。关键在于绕过“按文件轮转后批量分析”的传统做法，改为持续 tail 日志、逐行解析、内存中维护最近 60 秒的 IP 计数，并实时检测突增。
步骤一：稳定读取 Nginx 实时日志流
用 subprocess.Popen 启动 tail -F 命令，避免因日志轮转（logrotate）导致读取中断。需指定 --max-unchanged-stats=1 和重试逻辑，确保 inode 变化后自动 reconnect。
- 推荐日志格式为
log_format main '$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent" $request_time'; - 脚本启动时先检查日志文件是否存在，若被清空或重命名，自动寻找最新 access.log* 文件
- 每行用正则提取
$remote_addr（支持 IPv4/IPv6），忽略内部 IP（如 127.0.0.1、10.0.0.0/8）和健康检查 UA
步骤二：构建 60 秒滑动窗口计数器
不用固定 bucket，改用 collections.deque 存储（IP, 时间戳）二元组，配合字典做频次映射。每来一条日志，插入当前时间戳并清理队列头部超时项（ts ），同时更新 IP 计数。
- 每 1 秒触发一次统计快照：遍历当前计数字典，找出访问量 > 阈值（如 100 次/60秒）且较前 60 秒均值上涨 300% 的 IP
- 阈值可动态调整：初始设为 50，若过去 5 分钟每秒平均请求数
- 对高频 IP（如 CDN 回源）加白名单机制，通过配置文件或 Redis 快速匹配跳过告警
步骤三：轻量告警与低干扰通知
告警不走邮件或短信（延迟高），优先调用企业微信机器人 / 飞书 Webhook 推送结构化消息，含 Top3 异常 IP、QPS 估算、请求路径分布（取样前 5 条 $request）。
- 同一 IP 5 分钟内只告警一次，后续仅记录到本地 ring buffer（最多存 1000 条），供人工回溯
- 附带简单溯源建议：如该 IP 近 10 分钟 95% 请求为
GET /api/v1/user?token=xxx，提示可能为 token 暴力探测 - 支持 SIGUSR1 信号触发手动 dump 当前窗口数据到 JSON 文件，用于离线分析
运行与可靠性保障
用 systemd 托管进程，设置 Restart=on-failure 和内存限制（MemoryLimit=200M）。脚本自身做心跳检测：每 30 秒写一次时间戳到 /tmp/nginx-topip-heartbeat，监控脚本定期检查该文件是否更新。
- 日志解析失败行自动计入 error counter，连续 100 行失败则暂停 5 秒并 reload 正则模式
- 支持热重载配置：修改 YAML 配置后发送 SIGHUP，无需重启即可更新白名单、阈值策略、Webhook 地址
- 默认输出 INFO 级别到 stdout，DEBUG 级别可临时开启，便于调试正则匹配或时间窗口漂移问题
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
