企业官网建设流程全解析

痛点：问题爆发了，才知道不对劲
价值：在真正故障出现之前，系统已经提醒你了

很多线上事故在真正“炸掉”之前，日志里早就出现了异常信号：

• 某个错误关键词开始反复出现
• 某个模块的 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

🔍 与“普通日志监控”的根本区别

📌 一句话总结

真正有价值的日志监控，
不是等错误出现，
而是在错误“快要出现时”就提醒你。

👉 后续类似「日志感知 / 趋势预警 / 自动化工具」
会持续整理在《程序员自动化工具箱》。

如果你只想要
在事故之前，而不是事故之后才知道问题，
这个专栏你会用得上。