如何用脚本分析日志文件

本文目录导读：

1. 基础日志分析脚本
2. 高级功能脚本
3. 使用说明
4. 关键特性

我来介绍一个使用Python分析日志文件的综合脚本方案：

基础日志分析脚本

#!/usr/bin/env python3
"""
日志文件分析脚本
支持：访问日志、错误日志、自定义格式日志
"""
import re
import os
from collections import Counter, defaultdict
from datetime import datetime
import argparse
import json
class LogAnalyzer:
    def __init__(self, log_file):
        self.log_file = log_file
        self.lines = []
        self.stats = {
            'total_lines': 0,
            'error_count': 0,
            'warning_count': 0,
            'ip_addresses': Counter(),
            'status_codes': Counter(),
            'request_methods': Counter(),
            'hourly_distribution': Counter(),
            'error_messages': Counter()
        }
    def load_logs(self):
        """加载日志文件"""
        try:
            with open(self.log_file, 'r', encoding='utf-8', errors='ignore') as f:
                self.lines = f.readlines()
            self.stats['total_lines'] = len(self.lines)
            print(f"成功加载 {len(self.lines)} 行日志")
        except FileNotFoundError:
            print(f"错误：找不到日志文件 {self.log_file}")
            return False
        except Exception as e:
            print(f"错误：读取日志失败 - {e}")
            return False
        return True
    def parse_apache_log(self, line):
        """解析Apache/Nginx访问日志格式"""
        # 示例格式: 127.0.0.1 - - [10/Oct/2023:13:55:36] "GET /index.html HTTP/1.1" 200 2326
        pattern = r'(\d+\.\d+\.\d+\.\d+)\s+.*?\[(.*?)\]\s+"(\w+)\s+(.*?)\s+HTTP.*?"\s+(\d{3})\s+(\d+)'
        match = re.match(pattern, line)
        if match:
            ip, datetime_str, method, path, status, size = match.groups()
            return {
                'ip': ip,
                'timestamp': datetime_str,
                'method': method,
                'path': path,
                'status': status,
                'size': int(size)
            }
        return None
    def parse_error_log(self, line):
        """解析错误日志"""
        # 示例格式: [2023-10-10 13:55:36] [error] [client 127.0.0.1] File not found
        pattern = r'\[(.*?)\]\s+\[(\w+)\]\s+\[client\s+(.*?)\]\s+(.*)'
        match = re.match(pattern, line)
        if match:
            timestamp, level, client, message = match.groups()
            return {
                'timestamp': timestamp,
                'level': level,
                'client': client,
                'message': message
            }
        return None
    def analyze_access_log(self):
        """分析访问日志"""
        for line in self.lines:
            log = self.parse_apache_log(line)
            if log:
                self.stats['ip_addresses'][log['ip']] += 1
                self.stats['status_codes'][log['status']] += 1
                self.stats['request_methods'][log['method']] += 1
                # 按小时统计
                try:
                    hour = datetime.strptime(log['timestamp'][:14], '%d/%b/%Y:%H')
                    self.stats['hourly_distribution'][hour.strftime('%H')+':00'] += 1
                except:
                    pass
                # 统计错误
                if int(log['status']) >= 400:
                    self.stats['error_count'] += 1
                    self.stats['error_messages'][f"{log['status']} - {log['path']}"] += 1
    def analyze_error_log(self):
        """分析错误日志"""
        for line in self.lines:
            log = self.parse_error_log(line)
            if log:
                if log['level'] == 'error':
                    self.stats['error_count'] += 1
                    self.stats['error_messages'][log['message'][:100]] += 1
                elif log['level'] == 'warning':
                    self.stats['warning_count'] += 1
                self.stats['ip_addresses'][log['client']] += 1
    def generate_report(self):
        """生成分析报告"""
        report = []
        report.append("=" * 60)
        report.append("日志分析报告")
        report.append("=" * 60)
        report.append(f"分析时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append(f"日志文件：{self.log_file}")
        report.append(f"总行数：{self.stats['total_lines']}")
        report.append(f"错误数：{self.stats['error_count']}")
        report.append(f"警告数：{self.stats['warning_count']}")
        # IP 访问统计
        report.append("\n📊 访问量TOP 10 IP地址：")
        for ip, count in self.stats['ip_addresses'].most_common(10):
            report.append(f"  {ip}: {count}次")
        # HTTP 状态码
        if self.stats['status_codes']:
            report.append("\n📊 HTTP状态码分布：")
            total = sum(self.stats['status_codes'].values())
            for status, count in self.stats['status_codes'].most_common():
                percentage = (count / total) * 100
                report.append(f"  {status}: {count}次 ({percentage:.2f}%)")
        # 请求方法
        if self.stats['request_methods']:
            report.append("\n📊 请求方法分布：")
            for method, count in self.stats['request_methods'].most_common():
                report.append(f"  {method}: {count}次")
        # 按小时访问量
        if self.stats['hourly_distribution']:
            report.append("\n📊 每小时访问量：")
            for hour in sorted(self.stats['hourly_distribution'].keys()):
                count = self.stats['hourly_distribution'][hour]
                bar = '█' * (count // 10)
                report.append(f"  {hour}: {count}次 {bar}")
        # 错误详情
        if self.stats['error_messages']:
            report.append("\n❌ 错误的TOP 10：")
            for error, count in self.stats['error_messages'].most_common(10):
                report.append(f"  [{count}次] {error}")
        return '\n'.join(report)
    def save_report(self, output_file='report.txt'):
        """保存报告到文件"""
        report_content = self.generate_report()
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(report_content)
        print(f"报告已保存到 {output_file}")
    def export_json(self, output_file='report.json'):
        """导出JSON格式报告"""
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(dict(self.stats), f, ensure_ascii=False, indent=2)
        print(f"JSON报告已保存到 {output_file}")
def filter_logs_by_time(log_file, start_time, end_time):
    """按时间范围过滤日志"""
    filtered_lines = []
    pattern = r'\[(\d{2}/\w+/\d{4}:\d{2})'
    with open(log_file, 'r') as f:
        for line in f:
            match = re.search(pattern, line)
            if match:
                log_time = datetime.strptime(match.group(1), '%d/%b/%Y:%H')
                if start_time <= log_time <= end_time:
                    filtered_lines.append(line)
    return filtered_lines
def analyze_patterns(log_file, pattern):
    """搜索特定模式"""
    matched = []
    with open(log_file, 'r') as f:
        for line_num, line in enumerate(f, 1):
            if re.search(pattern, line, re.IGNORECASE):
                matched.append((line_num, line.rstrip()))
    return matched
def main():
    parser = argparse.ArgumentParser(description='日志分析工具')
    parser.add_argument('log_file', help='日志文件路径')
    parser.add_argument('--type', choices=['access', 'error'], default='access',
                       help='日志类型（访问日志/错误日志）')
    parser.add_argument('--search', help='搜索特定模式')
    parser.add_argument('--output', default='report.txt', help='输出报告文件')
    parser.add_argument('--json', action='store_true', help='导出JSON格式')
    args = parser.parse_args()
    # 创建分析器
    analyzer = LogAnalyzer(args.log_file)
    # 加载日志
    if not analyzer.load_logs():
        return
    # 搜索模式
    if args.search:
        print(f"\n🔍 搜索模式: {args.search}")
        matches = analyze_patterns(args.log_file, args.search)
        for line_num, line in matches[:20]:
            print(f"  第{line_num}行: {line}")
        print(f"共找到 {len(matches)} 处匹配")
        return
    # 分析日志
    if args.type == 'access':
        analyzer.analyze_access_log()
    else:
        analyzer.analyze_error_log()
    # 生成并保存报告
    report = analyzer.generate_report()
    print(report)
    analyzer.save_report(args.output)
    # 导出JSON
    if args.json:
        analyzer.export_json()
if __name__ == "__main__":
    main()

高级功能脚本

#!/usr/bin/env python3
"""高级日志分析功能"""
import asyncio
import aiofiles
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from collections import deque
import numpy as np
import time
class RealTimeLogMonitor:
    """实时日志监控"""
    def __init__(self, log_file, threshold=10):
        self.log_file = log_file
        self.threshold = threshold
        self.error_counts = deque(maxlen=100)
        self.last_position = 0
    async def tail_log(self, interval=1):
        """实时跟踪日志文件"""
        while True:
            try:
                async with aiofiles.open(self.log_file, 'r') as f:
                    await f.seek(self.last_position)
                    new_lines = await f.readlines()
                    self.last_position = await f.tell()
                    for line in new_lines:
                        yield line
            except Exception as e:
                print(f"监控异常: {e}")
            await asyncio.sleep(interval)
    def count_errors(self, line):
        """统计错误次数"""
        error_keywords = ['ERROR', 'FATAL', 'CRITICAL']
        for keyword in error_keywords:
            if keyword in line.upper():
                return True
        return False
    async def monitor(self):
        """启动监控"""
        error_count = 0
        start_time = time.time()
        print("🟢 实时监控启动...")
        async for line in self.tail_log():
            if self.count_errors(line):
                error_count += 1
                print(f"🔄 [{time.strftime('%H:%M:%S')}] 发现错误: {line.strip()}")
            if time.time() - start_time >= 60:  # 每分钟统计
                self.error_counts.append(error_count)
                print(f"📊 当前错误频率: {error_count}/分钟")
                if error_count > self.threshold:
                    print(f"⚠️ 警告！错误频率超过阈值 {self.threshold}")
                error_count = 0
                start_time = time.time()
class LogStatistics:
    """统计计算类"""
    @staticmethod
    def calculate_statistics(logs):
        """计算统计指标"""
        df = pd.DataFrame(logs)
        stats = {
            'total_requests': len(df),
            'unique_ips': df['ip'].nunique() if 'ip' in df else 0,
            'avg_response_size': df['size'].mean() if 'size' in df else 0,
            'max_response_size': df['size'].max() if 'size' in df else 0,
            'min_response_size': df['size'].min() if 'size' in df else 0,
            'std_deviation': df['size'].std() if 'size' in df else 0
        }
        return stats
    @staticmethod
    def time_series_analysis(logs, interval='1H'):
        """时间序列分析"""
        try:
            df = pd.DataFrame(logs)
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            df.set_index('timestamp', inplace=True)
            # 按时间间隔重采样
            resampled = df.resample(interval).count()
            return resampled
        except Exception as e:
            print(f"时间序列分析错误: {e}")
            return None
    @staticmethod
    def correlation_analysis(logs):
        """相关性分析"""
        try:
            df = pd.DataFrame(logs)
            df['status'] = df['status'].astype(int)
            df['size'] = df['size'].astype(int)
            # 计算相关性矩阵
            corr_matrix = df[['status', 'size']].corr()
            return corr_matrix
        except:
            return None
def visual_analysis(log_file):
    """可视化分析"""
    try:
        import pandas as pd
        # 读取并解析日志
        log_pattern = r'(\d+\.\d+\.\d+\.\d+)\s+.*?\[(.*?)\]\s+"(\w+)\s+(.*?)\s+HTTP.*?"\s+(\d{3})\s+(\d+)'
        data = []
        with open(log_file, 'r') as f:
            for line in f:
                match = re.match(log_pattern, line)
                if match:
                    ip, timestamp, method, path, status, size = match.groups()
                    data.append({
                        'ip': ip,
                        'timestamp': timestamp,
                        'method': method,
                        'path': path,
                        'status': int(status),
                        'size': int(size)
                    })
        if not data:
            print("没有可分析的数据")
            return
        # 创建DataFrame
        df = pd.DataFrame(data)
        # 创建图表
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        # 访问量趋势
        df['hour'] = df['timestamp'].str[14:16]
        hourly = df.groupby('hour').size()
        axes[0,0].plot(hourly.index, hourly.values, marker='o')
        axes[0,0].set_title('访问量时间趋势')
        axes[0,0].set_xlabel('小时')
        axes[0,0].set_ylabel('访问量')
        # 状态码分布
        status_counts = df['status'].value_counts().head(10)
        axes[0,1].pie(status_counts.values, labels=status_counts.index, autopct='%1.1f%%')
        axes[0,1].set_title('HTTP状态码分布')
        # IP访问TOP20
        ip_top = df['ip'].value_counts().head(20)
        axes[1,0].barh(range(len(ip_top)), ip_top.values)
        axes[1,0].set_yticks(range(len(ip_top)))
        axes[1,0].set_yticklabels(ip_top.index)
        axes[1,0].set_title('IP访问排行')
        # 响应大小分布
        axes[1,1].hist(df['size'], bins=50, alpha=0.7, color='skyblue', edgecolor='black')
        axes[1,1].set_title('响应大小分布')
        axes[1,1].set_xlabel('响应大小')
        axes[1,1].set_ylabel('频次')
        plt.tight_layout()
        plt.savefig('log_analysis.png', dpi=100)
        print("图表已保存: log_analysis.png")
        plt.show()
    except ImportError as e:
        print(f"需要安装相关依赖: {e}")
    except Exception as e:
        print(f"可视化分析错误: {e}")
def batch_processing(log_files):
    """批量处理多个日志文件"""
    all_stats = []
    for log_file in log_files:
        if not os.path.exists(log_file):
            print(f"跳过不存在的文件: {log_file}")
            continue
        analyzer = LogAnalyzer(log_file)
        if analyzer.load_logs():
            analyzer.analyze_access_log()
            stats = analyzer.stats
            stats['file'] = log_file
            all_stats.append(stats)
    # 汇总统计
    total_requests = sum(s['total_lines'] for s in all_stats)
    total_errors = sum(s['error_count'] for s in all_stats)
    print(f"\n{"="*40}")
    print(f"批量处理结果:")
    print(f"文件数量: {len(all_stats)}")
    print(f"总请求数: {total_requests}")
    print(f"总错误数: {total_errors}")
    return all_stats
# 使用示例
if __name__ == "__main__":
    # 示例1: 基础分析
    analyzer = LogAnalyzer('access.log')
    analyzer.load_logs()
    analyzer.analyze_access_log()
    report = analyzer.generate_report()
    print(report)
    # 示例2: 可视化
    visual_analysis('access.log')
    # 示例3: 实时监控
    # monitor = RealTimeLogMonitor('app.log')
    # asyncio.run(monitor.monitor())

使用说明

# 基本用法
python log_analyzer.py access.log
# 分析错误日志
python log_analyzer.py error.log --type error
# 搜索特定模式
python log_analyzer.py access.log --search "error|warning"
# 导出JSON格式
python log_analyzer.py access.log --json --output report.json
# 批量分析
python log_analyzer.py access.log error.log --output combined_report.txt

关键特性

1. 灵活解析：支持多种日志格式
2. 实时监控：可以监控日志文件变化
3. 统计分析：生成详细报告
4. 可视化：生成图表分析
5. 过滤搜索：按时间、模式过滤

这个脚本提供了全面的日志分析功能,可以根据具体需求调整和扩展。