痛点：问题爆发了，才知道不对劲
价值：在真正故障出现之前，系统已经提醒你了
很多线上事故在真正“炸掉”之前，日志里早就出现了异常信号：
- 某个错误关键词开始反复出现
- 某个模块的 WARN 数量持续上升
- 某些“看起来不严重”的词，频率突然异常
但绝大多数情况下，这些信息都被忽略了。
这篇文章直接做一件事：
用 Python 做一个「日志关键词雷达」，
持续扫描日志变化，在故障爆发前提前预警。
🎯 最终要实现的效果
系统持续运行后，你可以得到这样的能力：
- 实时统计关键日志词的出现频率
- 发现「短时间内异常增长」的关键词
- 自动输出预警报告（Markdown）
- 可作为定时任务 / 常驻进程运行
不是“等 ERROR 出现”，
而是趋势不对就已经被发现。
🧠 整体工作流程（直观版）
📦 项目结构
log-radar/ ├── radar.py ├── keywords.txt ├── alerts/ └── state.json🧩 关键词配置（keywords.txt）
关键词不追求多，而追求**“提前量”**：
timeout slow retry refused connection reset pool full queue full OOM GC overhead这些词的特点是：
- 很多时候还没 ERROR
- 但它们的频率变化极其敏感
⚙️ 核心逻辑说明（一句话）
不关心某个词出现了没有，
只关心：它是不是突然变多了。
🧱 核心代码（radar.py｜完整可运行）
#!/usr/bin/env python3# -*- coding: utf-8 -*-importjsonimporttimeimportargparsefromcollectionsimportdefaultdict,dequefromdatetimeimportdatetimefrompathlibimportPath WINDOW_SECONDS=300# 统计窗口：5 分钟THRESHOLD_MULTIPLIER=3# 增长倍数阈值defload_keywords(path):withopen(path,"r",encoding="utf-8")asf:return[line.strip().lower()forlineinfifline.strip()]defload_state(path):ifnotpath.exists():return{}returnjson.loads(path.read_text())defsave_state(path,state):path.write_text(json.dumps(state,indent=2,ensure_ascii=False))defnow_ts():returnint(time.time())defminute_key(ts):returndatetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M")defrender_alert_md(alerts):lines=[]lines.append("# 日志关键词雷达预警报告\n")lines.append(f"生成时间：{datetime.now()}\n")lines.append("| 关键词 | 最近窗口 | 历史基线 | 倍数 |")lines.append("|---|---:|---:|---:|")forainalerts:lines.append(f"| `{a['keyword']}` |{a['current']}|{a['baseline']}|{a['ratio']:.1f}x |")return"\n".join(lines)defmain():ap=argparse.ArgumentParser()ap.add_argument("--log",required=True,help="log file path")ap.add_argument("--keywords",default="keywords.txt")ap.add_argument("--state",default="state.json")ap.add_argument("--out",default="alerts")args=ap.parse_args()keywords=load_keywords(args.keywords)state_path=Path(args.state)out_dir=Path(args.out)out_dir.mkdir(exist_ok=True)state=load_state(state_path)counts=defaultdict(int)start_ts=now_ts()-WINDOW_SECONDSwithopen(args.log,"r",encoding="utf-8",errors="ignore")asf:forlineinf:line_l=line.lower()forkwinkeywords:ifkwinline_l:counts[kw]+=1alerts=[]forkw,curincounts.items():hist=state.get(kw,cur)ifhist>0andcur>=hist*THRESHOLD_MULTIPLIER:alerts.append({"keyword":kw,"current":cur,"baseline":hist,"ratio":cur/histifhistelse0,})state[kw]=int((hist*0.7)+(cur*0.3))save_state(state_path,state)ifalerts:ts=datetime.now().strftime("%Y%m%d_%H%M%S")md=render_alert_md(alerts)out_file=out_dir/f"alert_{ts}.md"out_file.write_text(md,encoding="utf-8")print(f"[ALERT] generated:{out_file}")else:print("[OK] no abnormal keyword growth detected")if__name__=="__main__":main()▶️ 运行方式
python radar.py --log app.log输出示例：
[ALERT] generated: alerts/alert_20260109_021500.md🧭 预警机制示意（时间维度）
📄 生成的预警报告示例（Markdown）
# 日志关键词雷达预警报告 | 关键词 | 最近窗口 | 历史基线 | 倍数 | |---|---:|---:|---:| | `timeout` | 42 | 10 | 4.2x | | `connection reset` | 15 | 4 | 3.7x |这份报告可以直接：
- 贴到 CSDN
- 发到群里
- 作为事故前置记录
⏰ 设置为周期运行
*/5 * * * * /usr/bin/python3 /opt/log-radar/radar.py\--log /var/log/app/app.log>>/var/log/log-radar.log2>&1🔍 与“普通日志监控”的根本区别
📌 一句话总结
真正有价值的日志监控，
不是等错误出现，
而是在错误“快要出现时”就提醒你。
👉 后续类似「日志感知 / 趋势预警 / 自动化工具」
会持续整理在《程序员自动化工具箱》。
如果你只想要
在事故之前，而不是事故之后才知道问题，
这个专栏你会用得上。
