如何写一个脚本统计分析日志

wen 实用脚本 2026-07-30 238

本文目录导读：

如何写一个脚本统计分析日志

基础日志分析脚本
使用方法
自定义分析场景
性能优化建议

我来为你创建一个日志统计分析脚本,这是一个通用且功能完整的示例：

基础日志分析脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
日志统计分析脚本
支持统计：访问次数、错误率、IP分布、状态码分布、时间趋势等
"""
import re
import sys
import os
from collections import Counter, defaultdict
from datetime import datetime, timedelta
import argparse
def parse_log_line(line):
    """解析单行日志，返回结构化数据"""
    # 通用日志格式（可根据实际情况调整）
    patterns = [
        # Nginx/Apache 标准格式
        r'(\d+\.\d+\.\d+\.\d+)\s+.*?\s+.*?\s+\[(.*?)\]\s+"(.*?)"\s+(\d+)\s+(\d+)\s+"(.*?)"\s+"(.*?)"',
        # 自定义格式：时间 IP 级别 消息
        r'(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+(\d+\.\d+\.\d+\.\d+)\s+(\w+)\s+(.*)',
    ]
    for pattern in patterns:
        match = re.match(pattern, line)
        if match:
            groups = match.groups()
            if len(groups) == 7:  # Nginx标准格式
                return {
                    'ip': groups[0],
                    'timestamp': groups[1],
                    'method': groups[2].split()[0] if groups[2] else '',
                    'url': groups[2].split()[1] if groups[2] else '',
                    'status_code': int(groups[3]),
                    'size': int(groups[4]),
                    'referer': groups[5],
                    'user_agent': groups[6]
                }
            elif len(groups) == 4:  # 自定义格式
                return {
                    'timestamp': groups[0],
                    'ip': groups[1],
                    'level': groups[2],
                    'message': groups[3]
                }
    return None
class LogAnalyzer:
    def __init__(self, log_file):
        self.log_file = log_file
        self.logs = []
        self.reset_stats()
    def reset_stats(self):
        """重置统计数据"""
        self.total_lines = 0
        self.parsed_lines = 0
        self.ip_counter = Counter()
        self.status_counter = Counter()
        self.hourly_counter = Counter()
        self.error_counter = Counter()
        self.method_counter = Counter()
        self.url_counter = Counter()
        self.level_counter = Counter()
    def parse_logs(self, sample_size=None):
        """解析日志文件"""
        print(f"正在解析日志文件: {self.log_file}")
        with open(self.log_file, 'r', encoding='utf-8', errors='ignore') as f:
            lines = f.readlines()
        if sample_size:
            lines = lines[:sample_size]
        self.total_lines = len(lines)
        for line in lines:
            parsed = parse_log_line(line.strip())
            if parsed:
                self.logs.append(parsed)
                self.update_stats(parsed)
                self.parsed_lines += 1
        print(f"完成解析: 总计{self.total_lines}行, 成功解析{self.parsed_lines}行")
    def update_stats(self, log_entry):
        """更新统计信息"""
        # IP统计
        if 'ip' in log_entry:
            self.ip_counter[log_entry['ip']] += 1
        # 状态码统计
        if 'status_code' in log_entry:
            self.status_counter[log_entry['status_code']] += 1
        # 小时统计
        if 'timestamp' in log_entry:
            try:
                hour = log_entry['timestamp'][:13]  # YYYY-MM-DD HH
                self.hourly_counter[hour] += 1
            except:
                pass
        # URL统计
        if 'url' in log_entry:
            self.url_counter[log_entry['url']] += 1
        # 方法统计
        if 'method' in log_entry:
            self.method_counter[log_entry['method']] += 1
        # 错误级别统计
        if 'level' in log_entry:
            self.level_counter[log_entry['level']] += 1
    def print_summary(self):
        """打印统计摘要"""
        print("\n" + "="*60)
        print("日志统计报告")
        print("="*60)
        # 基本信息
        print(f"\n📊 基本信息:")
        print(f"  总行数: {self.total_lines}")
        print(f"  解析成功: {self.parsed_lines}")
        print(f"  解析成功率: {self.parsed_lines/self.total_lines*100:.2f}%")
        # IP统计
        print(f"\n📍 IP地址统计 (Top 10):")
        for ip, count in self.ip_counter.most_common(10):
            print(f"  {ip}: {count} 次 ({count/self.parsed_lines*100:.1f}%)")
        # 状态码统计
        print(f"\n📡 HTTP状态码统计:")
        for code, count in sorted(self.status_counter.items()):
            # 分类状态码
            if code < 200:
                category = "信息"
            elif code < 300:
                category = "成功"
            elif code < 400:
                category = "重定向"
            elif code < 500:
                category = "客户端错误"
            else:
                category = "服务端错误"
            print(f"  {code} ({category}): {count} 次")
        # 错误率
        error_4xx = sum(c for k, c in self.status_counter.items() if 400 <= k < 500)
        error_5xx = sum(c for k, c in self.status_counter.items() if k >= 500)
        total_requests = sum(self.status_counter.values())
        if total_requests > 0:
            print(f"\n❌ 错误率分析:")
            print(f"  4xx错误(客户端): {error_4xx} ({error_4xx/total_requests*100:.1f}%)")
            print(f"  5xx错误(服务端): {error_5xx} ({error_5xx/total_requests*100:.1f}%)")
            print(f"  总错误率: {(error_4xx+error_5xx)/total_requests*100:.1f}%")
        # 时间分布
        if self.hourly_counter:
            print(f"\n🕐 时间分布 (Top 10时间段):")
            for hour, count in self.hourly_counter.most_common(10):
                print(f"  {hour}: {count} 次")
        # URL统计
        print(f"\n🔗 URL访问统计 (Top 10):")
        for url, count in self.url_counter.most_common(10):
            print(f"  {url[:50]}: {count} 次")
        # 方法统计
        if self.method_counter:
            print(f"\n⚙️ HTTP方法统计:")
            for method, count in self.method_counter.most_common():
                print(f"  {method}: {count} 次")
        # 错误级别统计（应用日志）
        if self.level_counter:
            print(f"\n⚠️ 日志级别统计:")
            for level, count in self.level_counter.most_common():
                print(f"  {level}: {count} 次")
        # 异常检测
        self.detect_anomalies()
    def detect_anomalies(self):
        """检测异常行为"""
        print(f"\n🚨 异常检测:")
        # 检测高频访问IP（DDoS可能）
        threshold = max(10, int(self.parsed_lines * 0.05))
        high_freq_ips = [(ip, count) for ip, count in self.ip_counter.items() if count > threshold]
        if high_freq_ips:
            print(f"  ⚠️ 高频访问IP (>{threshold}次):")
            for ip, count in high_freq_ips[:5]:
                print(f"    {ip}: {count} 次")
        # 检测404错误
        _404_urls = [(url, count) for url, code in self.url_counter.items() if code == 404]
        if _404_urls:
            print(f"  404错误URL (Top 5):")
            for url, count in self.url_counter.most_common(5):
                pass
    def export_report(self, output_file):
        """导出统计报告"""
        import json
        report = {
            'generated_at': datetime.now().isoformat(),
            'total_lines': self.total_lines,
            'parsed_lines': self.parsed_lines,
            'ip_stats': dict(self.ip_counter.most_common(20)),
            'status_stats': dict(self.status_counter),
            'url_stats': dict(self.url_counter.most_common(20)),
            'hourly_stats': dict(sorted(self.hourly_counter.items())),
        }
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
        print(f"\n报告已导出: {output_file}")
def main():
    parser = argparse.ArgumentParser(description='日志统计分析工具')
    parser.add_argument('logfile', help='日志文件路径')
    parser.add_argument('-n', '--lines', type=int, help='分析行数限制')
    parser.add_argument('-o', '--output', help='导出报告文件路径')
    parser.add_argument('-s', '--show', action='store_true', help='显示详细统计')
    args = parser.parse_args()
    if not os.path.exists(args.logfile):
        print(f"错误: 文件 {args.logfile} 不存在")
        sys.exit(1)
    analyzer = LogAnalyzer(args.logfile)
    analyzer.parse_logs(sample_size=args.lines)
    analyzer.print_summary()
    if args.output:
        analyzer.export_report(args.output)
if __name__ == "__main__":
    main()

使用方法

基本用法

# 分析整个日志文件
python log_analyzer.py /var/log/nginx/access.log
# 只分析前10000行
python log_analyzer.py access.log -n 10000
# 导出JSON报告
python log_analyzer.py access.log -o report.json
# 显示详细统计
python log_analyzer.py access.log -s

简化版脚本（适合快速使用）

#!/bin/bash
# quick_log_stats.sh - 快速日志统计脚本
# 使用说明
if [ $# -eq 0 ]; then
    echo "用法: $0 <日志文件>"
    exit 1
fi
LOG_FILE=$1
echo "════════════════════════════════════════"
echo "日志统计报告 - $(date)"
echo "════════════════════════════════════════"
# 统计基本信息
echo ""
echo "📊 基本信息"
echo "总行数: $(wc -l < $LOG_FILE)"
echo "文件大小: $(du -h $LOG_FILE | cut -f1)"
# IP统计
echo ""
echo "📍 IP统计 (Top 10)"
awk '{print $1}' $LOG_FILE | sort | uniq -c | sort -rn | head -10
# 状态码统计
echo ""
echo "📡 HTTP状态码统计"
awk '{print $9}' $LOG_FILE | sort | uniq -c | sort -rn
# 404错误统计
echo ""
echo "❌ 404错误URL"
grep " 404 " $LOG_FILE | awk '{print $7}' | sort | uniq -c | sort -rn | head -10
# 每小时请求量
echo ""
echo "🕐 每小时请求量"
awk '{print $4}' $LOG_FILE | cut -d: -f1,2 | sort | uniq -c | sort -rn | head -20

自定义分析场景

分析特定时间段

def analyze_time_range(logs, start_time, end_time):
    """分析特定时间段的日志"""
    filtered_logs = []
    for log in logs:
        if 'timestamp' in log:
            try:
                log_time = datetime.strptime(log['timestamp'], '%Y-%m-%d %H:%M:%S')
                if start_time <= log_time <= end_time:
                    filtered_logs.append(log)
            except:
                pass
    return filtered_logs

错误日志分析

def analyze_errors(logs):
    """分析错误日志"""
    errors = [log for log in logs if log.get('level') in ['ERROR', 'CRITICAL']]
    print(f"错误总数: {len(errors)}")
    # 按错误消息分组
    error_messages = Counter()
    for error in errors:
        error_messages[error.get('message', '')] += 1
    for msg, count in error_messages.most_common(10):
        print(f"  [{count}] {msg[:100]}")