最稳方案是正则+pandas本地解析access.log:预编译正则抽remote_addr,清洗ip(strip、去空、滤ipv6),按阈值+状态码分布(4xx/5xx>60%)+敏感路径+响应时间多维识别异常ip,流式构建字典转dataframe并显式转换status/req_time类型,严格限定时间窗口过滤。

直接用正则 + pandas 统计 IP 出现频次最稳
别一上来就上 Redis 或 Flink,日常运维中 95% 的高频 IP 识别需求靠本地解析就能搞定。关键不是“快”,而是“准”——尤其要避开日志格式不一致、空行、时间偏移、IPv6 混入等实际坑点。
核心逻辑就是:逐行匹配 access.log,抽 remote_addr 字段,统计频次,再按阈值过滤。示例正则必须适配你真实的 log_format,比如你的配置含 $http_x_forwarded_for,就不能只信 $remote_addr。
- 用
re.compile预编译正则,避免每行都重编译 - 加
try/except跳过解析失败的行(常见于日志轮转中间态或错误写入) - 对
ip字段做基础清洗:strip()、过滤掉-和空字符串、跳过 IPv6(除非你明确需要) - 不要直接用
line.split()[0]——Nginx 日志里空格可能出现在 User-Agent 里,会错位
怎么定义“异常 IP”?得结合状态码和请求路径
只看访问次数是危险的。一个爬虫每秒刷 10 次 /robots.txt,次数高但危害低;另一个 IP 在 2 分钟内发了 87 次 POST /login 且 92% 返回 401 或 403,这才是真异常。
所以统计完频次后,必须叠加至少两个维度:
-
status分布:筛选出4xx或5xx占比 > 60% 的 IP -
url模式:匹配高频出现的敏感路径,如/wp-login.php、/api/v1/auth、/admin/ -
request_time异常:平均响应时间可能是探测,<code>> 5s可能是慢速攻击
示例判断逻辑:if ip_counts[ip] > 100 and status_4xx_pct[ip] > 0.7 and any(pattern in url for pattern in ['/login', '/auth']):
用 pandas 做分组聚合时注意 dtype 和内存
日志量大时(单日 > 500MB),直接 pd.read_csv 会爆内存或报错。正确做法是流式解析 + 构建字典,再一次性转 DataFrame。
更关键的是字段类型:默认所有列都是 object,status 算不了均值,req_time 排不了序。必须显式转换:
-
df['status'] = pd.to_numeric(df['status'], errors='coerce')(errors='coerce'把非法值变NaN) df['req_time'] = pd.to_numeric(df['req_time'], errors='coerce')-
df = df.dropna(subset=['status', 'req_time'])再继续分析
否则 df.groupby('ip')['status'].mean() 会静默返回 NaN,你还以为没数据。
输出结果前务必校验时间范围
自动分析脚本最容易被忽略的点:没限定时间窗口。你跑的是昨天的日志,但脚本默认读全量 access.log(可能包含上周轮转的压缩包内容),或者时间字段解析失败导致 time_local 全是 NaT,最后统计的“高频 IP”其实是历史噪音。
解决方式很简单:
- 加参数
--since "2026-04-12 00:00:00",解析后用df[df['time'] >= since_dt]过滤 - 如果日志文件名带日期(如
access.log-20260412.gz),优先从文件名取时间,比解析每行time_local更可靠 - 用
gzip.open直接读压缩日志,别先解压——省 IO,也避免磁盘满
没有时间约束的“高频 IP”列表,和没过滤的 SELECT * 一样,看着热闹,用起来全是坑。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











