116 lines
3.2 KiB
Python
116 lines
3.2 KiB
Python
|
#!/usr/bin/env python
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
简单日志汇总脚本
|
|||
|
|
|
|||
|
|
遍历 logs/ 目录下最近的若干个 dashboard.log 文件，统计 ERROR / WARNING / CRITICAL，
|
|||
|
|
并输出简要汇总信息，供 log-summary Skill 调用。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import re
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import List, Tuple
|
|||
|
|
|
|||
|
|
|
|||
|
|
LOG_ROOT = Path("logs")
|
|||
|
|
LOG_FILENAME = "dashboard.log"
|
|||
|
|
MAX_FILES = 5 # 最多分析最近 N 个日志文件
|
|||
|
|
|
|||
|
|
|
|||
|
|
LEVEL_PATTERNS = {
|
|||
|
|
"ERROR": re.compile(r"\bERROR\b"),
|
|||
|
|
"WARNING": re.compile(r"\bWARNING\b"),
|
|||
|
|
"CRITICAL": re.compile(r"\bCRITICAL\b"),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def find_log_files() -> List[Path]:
|
|||
|
|
if not LOG_ROOT.exists():
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
candidates: List[Tuple[float, Path]] = []
|
|||
|
|
for root, dirs, files in os.walk(LOG_ROOT):
|
|||
|
|
if LOG_FILENAME in files:
|
|||
|
|
p = Path(root) / LOG_FILENAME
|
|||
|
|
try:
|
|||
|
|
mtime = p.stat().st_mtime
|
|||
|
|
except OSError:
|
|||
|
|
continue
|
|||
|
|
candidates.append((mtime, p))
|
|||
|
|
|
|||
|
|
# 按修改时间从新到旧排序
|
|||
|
|
candidates.sort(key=lambda x: x[0], reverse=True)
|
|||
|
|
return [p for _, p in candidates[:MAX_FILES]]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def summarize_file(path: Path):
|
|||
|
|
counts = {level: 0 for level in LEVEL_PATTERNS.keys()}
|
|||
|
|
top_messages = {}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with path.open("r", encoding="utf-8", errors="ignore") as f:
|
|||
|
|
for line in f:
|
|||
|
|
for level, pattern in LEVEL_PATTERNS.items():
|
|||
|
|
if pattern.search(line):
|
|||
|
|
counts[level] += 1
|
|||
|
|
# 取日志消息部分做前缀（粗略）
|
|||
|
|
msg = line.strip()
|
|||
|
|
# 截断以防过长
|
|||
|
|
msg = msg[:200]
|
|||
|
|
top_messages[msg] = top_messages.get(msg, 0) + 1
|
|||
|
|
break
|
|||
|
|
except OSError as e:
|
|||
|
|
print(f"[!] 读取日志失败 {path}: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# 取 Top 5
|
|||
|
|
top_list = sorted(top_messages.items(), key=lambda x: x[1], reverse=True)[:5]
|
|||
|
|
return counts, top_list
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
log_files = find_log_files()
|
|||
|
|
if not log_files:
|
|||
|
|
print("未找到任何日志文件（logs/*/dashboard.log）。")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
print(f"共找到 {len(log_files)} 个最近的日志文件（最多 {MAX_FILES} 个）：\n")
|
|||
|
|
|
|||
|
|
overall = {level: 0 for level in LEVEL_PATTERNS.keys()}
|
|||
|
|
|
|||
|
|
for idx, path in enumerate(log_files, start=1):
|
|||
|
|
print(f"[{idx}] 日志文件: {path}")
|
|||
|
|
result = summarize_file(path)
|
|||
|
|
if result is None:
|
|||
|
|
print(" 无法读取该日志文件。\n")
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
counts, top_list = result
|
|||
|
|
for level, c in counts.items():
|
|||
|
|
overall[level] += c
|
|||
|
|
print(
|
|||
|
|
" 级别统计: "
|
|||
|
|
+ ", ".join(f"{lvl}={counts[lvl]}" for lvl in LEVEL_PATTERNS.keys())
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
if top_list:
|
|||
|
|
print(" Top 错误/警告消息：")
|
|||
|
|
for msg, n in top_list:
|
|||
|
|
print(f" [{n}次] {msg}")
|
|||
|
|
else:
|
|||
|
|
print(" 未发现 ERROR/WARNING/CRITICAL 级别日志。")
|
|||
|
|
|
|||
|
|
print()
|
|||
|
|
|
|||
|
|
print("总体统计：")
|
|||
|
|
print(
|
|||
|
|
" "
|
|||
|
|
+ ", ".join(f"{lvl}={overall[lvl]}" for lvl in LEVEL_PATTERNS.keys())
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|
|||
|
|
|