如何写分类统计报错日志脚本

wen 实用脚本 2026-06-26 222

本文目录导读：

如何写分类统计报错日志脚本

方案一：Python脚本（推荐，功能最完整）
方案二：Shell脚本（轻量级，适合Linux环境）
方案三：使用awk的快速统计脚本
使用示例
高级功能扩展

我来提供一个实用的分类统计报错日志脚本,包含多种实现方案：

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