本文目录导读：
我来提供一个实用的分类统计报错日志脚本,包含多种实现方案：
Python脚本（推荐，功能最完整）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
日志错误分类统计脚本
功能：读取日志文件，按错误类型进行分类统计
"""
import re
import os
from collections import defaultdict
from datetime import datetime
import argparse
class LogErrorAnalyzer:
def __init__(self, log_file):
self.log_file = log_file
self.error_stats = defaultdict(lambda: {
'count': 0,
'details': [],
'first_seen': None,
'last_seen': None,
'severity': 'ERROR'
})
def analyze(self):
"""分析日志文件"""
if not os.path.exists(self.log_file):
print(f"错误：文件 {self.log_file} 不存在")
return False
print(f"开始分析日志文件：{self.log_file}")
print(f"分析时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("-" * 60)
try:
with open(self.log_file, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
self._process_line(line.strip(), line_num)
self._generate_report()
return True
except Exception as e:
print(f"读取文件错误：{e}")
return False
def _process_line(self, line, line_num):
"""处理每一行日志"""
if not line:
return
# 识别的错误模式
error_patterns = {
'CONNECTION_ERROR': [r'connection.*(?:refused|timeout|reset)',
r'connect.*failed'],
'AUTH_ERROR': [r'(?:login|auth|authenticate).*(?:fail|error|denied)',
r'invalid.*(?:password|token|credential)'],
'TIMEOUT_ERROR': [r'timeout', r'timed out'],
'DATABASE_ERROR': [r'(?:database|db|sql).*(?:error|fail|exception)',
r'query.*(?:failed|error)'],
'MEMORY_ERROR': [r'(?:memory|out of memory|oom)',
r'memory.*(?:overflow|limit)'],
'FILE_ERROR': [r'(?:file|io).*(?:not found|error|denied)',
r'permission.*denied'],
'NETWORK_ERROR': [r'(?:network|socket).*(?:error|fail)',
r'dns.*(?:error|fail)'],
'VALIDATION_ERROR': [r'(?:validation|invalid|illegal).*(?:argument|parameter|input)',
r'value.*(?:out of range|invalid)'],
'RUNTIME_ERROR': [r'(?:runtime|exception|error).*(?:occurred|raised)',
r'unexpected.*(?:error|exception)'],
}
# 通用错误匹配（如果没有更具体的分类）
is_error = False
for category, patterns in error_patterns.items():
for pattern in patterns:
if re.search(pattern, line, re.IGNORECASE):
self._record_error(category, line, line_num)
is_error = True
break
if is_error:
break
# 如果包含 ERROR 或 Exception 但没有匹配到具体分类
if not is_error and re.search(r'(?:ERROR|Exception|FATAL)', line, re.IGNORECASE):
self._record_error('OTHER_ERROR', line, line_num)
def _record_error(self, category, line, line_num):
"""记录错误信息"""
timestamp = self._extract_timestamp(line)
self.error_stats[category]['count'] += 1
# 记录详细信息和时间
detail = {
'line': line_num,
'message': line[:200], # 截取前200字符
'timestamp': timestamp or datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
if len(self.error_stats[category]['details']) < 100: # 最多保存100条详细记录
self.error_stats[category]['details'].append(detail)
if not self.error_stats[category]['first_seen'] or timestamp:
self.error_stats[category]['first_seen'] = detail['timestamp']
self.error_stats[category]['last_seen'] = detail['timestamp']
def _extract_timestamp(self, line):
"""尝试从日志行中提取时间戳"""
patterns = [
r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}',
r'\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2}',
r'\d{2}-\d{2}-\d{4} \d{2}:\d{2}:\d{2}',
]
for pattern in patterns:
match = re.search(pattern, line)
if match:
return match.group()
return None
def _generate_report(self):
"""生成统计报告"""
if not self.error_stats:
print("未发现任何错误日志")
return
total_errors = sum(v['count'] for v in self.error_stats.values())
print(f"\n{'='*60}")
print(f"错误分类统计报告")
print(f"{'='*60}")
print(f"总错误数：{total_errors}")
print(f"错误类型数：{len(self.error_stats)}")
print(f"{'='*60}")
# 按错误数量降序排列
sorted_errors = sorted(self.error_stats.items(),
key=lambda x: x[1]['count'], reverse=True)
for category, stats in sorted_errors:
percentage = (stats['count'] / total_errors) * 100
print(f"\n📊 {self._format_category(category)}")
print(f" 数量：{stats['count']} ({percentage:.2f}%)")
print(f" 首次时间：{stats['first_seen'] or 'N/A'}")
print(f" 最后时间：{stats['last_seen'] or 'N/A'}")
# 显示前5条详细记录
if stats['details']:
print(f" 最近错误示例（最多显示5条）：")
for detail in stats['details'][-5:]:
print(f" • 第{detail['line']}行: {detail['message'][:100]}")
def _format_category(self, category):
"""格式化错误类别名称"""
names = {
'CONNECTION_ERROR': '连接错误',
'AUTH_ERROR': '认证错误',
'TIMEOUT_ERROR': '超时错误',
'DATABASE_ERROR': '数据库错误',
'MEMORY_ERROR': '内存错误',
'FILE_ERROR': '文件错误',
'NETWORK_ERROR': '网络错误',
'VALIDATION_ERROR': '验证错误',
'RUNTIME_ERROR': '运行时错误',
'OTHER_ERROR': '其他错误'
}
return f"{names.get(category, category)} ({category})"
def main():
parser = argparse.ArgumentParser(description='日志错误分类统计工具')
parser.add_argument('log_file', help='要分析的日志文件路径')
parser.add_argument('--output', '-o', help='输出报告到文件（可选）')
args = parser.parse_args()
analyzer = LogErrorAnalyzer(args.log_file)
analyzer.analyze()
if args.output and hasattr(analyzer, 'error_stats'):
with open(args.output, 'w', encoding='utf-8') as f:
f.write("错误统计报告\n")
f.write(f"文件: {args.log_file}\n")
f.write(f"时间: {datetime.now()}\n\n")
for category, stats in analyzer.error_stats.items():
f.write(f"{category}: {stats['count']}次\n")
if __name__ == "__main__":
main()
Shell脚本（轻量级，适合Linux环境）
#!/bin/bash
# 日志错误分类统计脚本
# 使用方法: ./error_stats.sh [日志文件路径]
LOG_FILE="${1:-/var/log/syslog}"
if [ ! -f "$LOG_FILE" ]; then
echo "错误：文件 $LOG_FILE 不存在"
exit 1
fi
echo "=============================="
echo "日志错误分类统计"
echo "文件: $LOG_FILE"
echo "时间: $(date)"
echo "=============================="
# 定义错误类别和对应的grep模式
declare -A ERROR_CATEGORIES
ERROR_CATEGORIES=(
["连接错误"]="connection refused|connection timeout|connection reset"
["认证错误"]="authentication failed|login failed|permission denied"
["超时错误"]="timeout|timed out"
["数据库错误"]="database error|SQL error|query failed"
["内存错误"]="out of memory|memory error|OOM"
["文件错误"]="file not found|IO error|open failed"
["网络错误"]="network error|socket error|connection failed"
["系统错误"]="kernel error|fatal error|segfault"
["其他错误"]="ERROR|FATAL|Exception"
)
# 统计所有错误
total_errors=0
for category in "${!ERROR_CATEGORIES[@]}"; do
pattern="${ERROR_CATEGORIES[$category]}"
count=$(grep -icE "$pattern" "$LOG_FILE" 2>/dev/null)
if [ "$count" -gt 0 ]; then
total_errors=$((total_errors + count))
echo ""
echo "📊 $category"
echo " 数量: $count"
# 显示最近的错误示例（最多3条）
if [ "$count" -le 3 ]; then
echo " 错误示例:"
grep -iE "$pattern" "$LOG_FILE" | head -3 | while IFS= read -r line; do
echo " • ${line:0:100}"
done
else
echo " 最近错误示例:"
grep -iE "$pattern" "$LOG_FILE" | tail -3 | while IFS= read -r line; do
echo " • ${line:0:100}"
done
fi
fi
done
echo ""
echo "=============================="
echo "统计总计"
echo "总错误数: $total_errors"
echo "=============================="
使用awk的快速统计脚本
#!/bin/bash
# 使用awk快速统计错误类型
log_file="${1:-/var/log/syslog}"
if [ ! -f "$log_file" ]; then
echo "文件不存在: $log_file"
exit 1
fi
awk '
BEGIN {
print "错误类型统计报告"
print "================\n"
}
{
# 初始化错误计数
if ($0 ~ /[Ee]rror/) total_errors++
# 分类统计
if ($0 ~ /connection.*refused|connection.*timeout/) connection_error++
if ($0 ~ /login.*failed|auth.*fail/) auth_error++
if ($0 ~ /[Tt]imeout/) timeout_error++
if ($0 ~ /database.*error|SQL.*error/) database_error++
if ($0 ~ /out of memory|memory.*error/) memory_error++
if ($0 ~ /file not found|IO error/) file_error++
if ($0 ~ /network.*error|socket.*error/) network_error++
}
END {
printf "总错误行数: %d\n\n", total_errors
printf "连接错误: %d\n", connection_error + 0
printf "认证错误: %d\n", auth_error + 0
printf "超时错误: %d\n", timeout_error + 0
printf "数据库错误: %d\n", database_error + 0
printf "内存错误: %d\n", memory_error + 0
printf "文件错误: %d\n", file_error + 0
printf "网络错误: %d\n", network_error + 0
print "\n========= 统计完成 ========="
}' "$log_file"
使用示例
# Python版本 python3 log_error_analyzer.py /var/log/app.log python3 log_error_analyzer.py /var/log/app.log --output report.txt # Shell版本 chmod +x error_stats.sh ./error_stats.sh /var/log/syslog # awk版本 ./error_analysis.awk /var/log/application.log
高级功能扩展
如果你需要更强大的功能,可以考虑：
- 实时监控：使用
tail -f实时分析日志 - 邮件告警：当某类错误超过阈值时发送通知
- 趋势分析：按时间段统计错误趋势
- JSON输出：方便集成到监控系统
- 多文件分析：同时分析多个日志文件
这个脚本可以根据你的具体需求进行调整,比如修改错误模式匹配规则、添加更多错误类别等。
