pip install pandas matplotlib numpy tqdm python-dateutil

python
 
import re
import os
import gzip
import pandas as pd
import numpy as np
from datetime import datetime
from dateutil import parser
from tqdm import tqdm
import matplotlib.pyplot as plt
from typing import List, Dict, Tuple, Optional, Pattern
 
class LogAnalyzer:
    """專(zhuān)業(yè)的日志分析工具"""
    
    DEFAULT_PATTERNS = {
        'timestamp': r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})',
        'level': r'(?P<level>DEBUG|INFO|WARNING|ERROR|CRITICAL)',
        'message': r'(?P<message>.*)',
        'source': r'(?P<source>\w+\.\w+)'
    }
    
    def __init__(self, log_dir: str, output_dir: str = "log_analysis"):
        """
        初始化日志分析器
        
        :param log_dir: 日志目錄路徑
        :param output_dir: 輸出目錄路徑
        """
        self.log_dir = log_dir
        self.output_dir = output_dir
        os.makedirs(self.output_dir, exist_ok=True)
        
        # 編譯正則表達(dá)式
        self.patterns = {
            name: re.compile(pattern) 
            for name, pattern in self.DEFAULT_PATTERNS.items()
        }
        
        # 分析結(jié)果存儲(chǔ)
        self.stats = {
            'total_lines': 0,
            'level_counts': {},
            'source_counts': {},
            'errors': [],
            'timeline': []
        }
    
    def detect_log_format(self, sample_lines: List[str]) -> bool:
        """自動(dòng)檢測(cè)日志格式"""
        for line in sample_lines[:10]:  # 檢查前10行
            match = self._parse_line(line)
            if not match:
                return False
        return True
    
    def _parse_line(self, line: str) -> Optional[Dict[str, str]]:
        """解析單行日志"""
        combined_pattern = re.compile(
            r'^{timestamp}\s+{level}\s+\[{source}\]\s+{message}$'.format(
                **self.DEFAULT_PATTERNS
            )
        )
        
        match = combined_pattern.match(line.strip())
        if match:
            return match.groupdict()
        return None
    
    def _read_log_file(self, filepath: str) -> List[str]:
        """讀取日志文件，支持gzip壓縮格式"""
        if filepath.endswith('.gz'):
            with gzip.open(filepath, 'rt', encoding='utf-8') as f:
                return f.readlines()
        else:
            with open(filepath, 'r', encoding='utf-8') as f:
                return f.readlines()
    
    def analyze_file(self, filepath: str):
        """分析單個(gè)日志文件"""
        lines = self._read_log_file(filepath)
        filename = os.path.basename(filepath)
        
        for line in tqdm(lines, desc=f"分析 {filename}"):
            self.stats['total_lines'] += 1
            parsed = self._parse_line(line)
            
            if not parsed:
                continue  # 跳過(guò)無(wú)法解析的行
                
            # 更新時(shí)間線數(shù)據(jù)
            try:
                dt = parser.parse(parsed['timestamp'])
                self.stats['timeline'].append({
                    'timestamp': dt,
                    'level': parsed['level'],
                    'source': parsed['source']
                })
            except (ValueError, KeyError):
                pass
            
            # 統(tǒng)計(jì)日志級(jí)別
            level = parsed.get('level', 'UNKNOWN')
            self.stats['level_counts'][level] = self.stats['level_counts'].get(level, 0) + 1
            
            # 統(tǒng)計(jì)來(lái)源
            source = parsed.get('source', 'unknown')
            self.stats['source_counts'][source] = self.stats['source_counts'].get(source, 0) + 1
            
            # 記錄錯(cuò)誤信息
            if level in ('ERROR', 'CRITICAL'):
                self.stats['errors'].append({
                    'timestamp': parsed.get('timestamp'),
                    'source': source,
                    'message': parsed.get('message', '')[:500]  # 截?cái)嚅L(zhǎng)消息
                })
    
    def analyze_directory(self):
        """分析目錄下所有日志文件"""
        log_files = []
        for root, _, files in os.walk(self.log_dir):
            for file in files:
                if file.endswith(('.log', '.txt', '.gz')):
                    log_files.append(os.path.join(root, file))
        
        print(f"發(fā)現(xiàn) {len(log_files)} 個(gè)日志文件待分析...")
        for filepath in log_files:
            self.analyze_file(filepath)
    
    def generate_reports(self):
        """生成分析報(bào)告"""
        # 準(zhǔn)備時(shí)間序列數(shù)據(jù)
        timeline_df = pd.DataFrame(self.stats['timeline'])
        timeline_df.set_index('timestamp', inplace=True)
        
        # 1. 生成日志級(jí)別分布圖
        self._plot_level_distribution()
        
        # 2. 生成時(shí)間序列圖
        self._plot_timeline(timeline_df)
        
        # 3. 生成錯(cuò)誤報(bào)告
        self._generate_error_report()
        
        # 4. 保存統(tǒng)計(jì)結(jié)果
        self._save_statistics()
    
    def _plot_level_distribution(self):
        """繪制日志級(jí)別分布圖"""
        levels = list(self.stats['level_counts'].keys())
        counts = list(self.stats['level_counts'].values())
        
        plt.figure(figsize=(10, 6))
        bars = plt.bar(levels, counts, color=['green', 'blue', 'orange', 'red', 'purple'])
        
        # 添加數(shù)值標(biāo)簽
        for bar in bars:
            height = bar.get_height()
            plt.text(bar.get_x() + bar.get_width()/2., height,
                    f'{height:,}', ha='center', va='bottom')
        
        plt.title('日志級(jí)別分布')
        plt.xlabel('日志級(jí)別')
        plt.ylabel('出現(xiàn)次數(shù)')
        plt.grid(axis='y', linestyle='--', alpha=0.7)
        
        # 保存圖片
        output_path = os.path.join(self.output_dir, 'level_distribution.png')
        plt.savefig(output_path, bbox_inches='tight', dpi=300)
        plt.close()
        print(f"已保存日志級(jí)別分布圖: {output_path}")
    
    def _plot_timeline(self, df: pd.DataFrame):
        """繪制時(shí)間序列圖"""
        plt.figure(figsize=(14, 8))
        
        # 按小時(shí)重采樣
        hourly = df.groupby([pd.Grouper(freq='H'), 'level']).size().unstack()
        hourly.plot(kind='area', stacked=True, alpha=0.7, figsize=(14, 8))
        
        plt.title('日志活動(dòng)時(shí)間線（按小時(shí)）')
        plt.xlabel('時(shí)間')
        plt.ylabel('日志數(shù)量')
        plt.grid(True, linestyle='--', alpha=0.5)
        plt.legend(title='日志級(jí)別')
        
        # 保存圖片
        output_path = os.path.join(self.output_dir, 'activity_timeline.png')
        plt.savefig(output_path, bbox_inches='tight', dpi=300)
        plt.close()
        print(f"已保存活動(dòng)時(shí)間線圖: {output_path}")
    
    def _generate_error_report(self):
        """生成錯(cuò)誤報(bào)告"""
        if not self.stats['errors']:
            print("未發(fā)現(xiàn)錯(cuò)誤日志")
            return
            
        df = pd.DataFrame(self.stats['errors'])
        
        # 按錯(cuò)誤源分組統(tǒng)計(jì)
        error_stats = df.groupby('source').size().sort_values(ascending=False)
        
        # 保存CSV
        csv_path = os.path.join(self.output_dir, 'error_report.csv')
        df.to_csv(csv_path, index=False, encoding='utf-8-sig')
        
        # 生成錯(cuò)誤源分布圖
        plt.figure(figsize=(12, 6))
        error_stats.plot(kind='bar', color='coral')
        plt.title('錯(cuò)誤來(lái)源分布')
        plt.xlabel('來(lái)源組件')
        plt.ylabel('錯(cuò)誤數(shù)量')
        plt.grid(axis='y', linestyle='--', alpha=0.7)
        
        img_path = os.path.join(self.output_dir, 'error_source_distribution.png')
        plt.savefig(img_path, bbox_inches='tight', dpi=300)
        plt.close()
        
        print(f"已生成錯(cuò)誤報(bào)告:\n- CSV文件: {csv_path}\n- 分布圖: {img_path}")
    
    def _save_statistics(self):
        """保存統(tǒng)計(jì)結(jié)果"""
        stats_path = os.path.join(self.output_dir, 'summary_statistics.txt')
        
        with open(stats_path, 'w', encoding='utf-8') as f:
            f.write("=== 日志分析摘要 ===\n\n")
            f.write(f"分析時(shí)間: {datetime.now().isoformat()}\n")
            f.write(f"日志目錄: {self.log_dir}\n")
            f.write(f"分析日志行數(shù): {self.stats['total_lines']:,}\n\n")
            
            f.write("日志級(jí)別統(tǒng)計(jì):\n")
            for level, count in sorted(self.stats['level_counts'].items()):
                f.write(f"- {level}: {count:,} ({count/self.stats['total_lines']:.1%})\n")
            
            f.write("\n來(lái)源組件統(tǒng)計(jì) (Top 10):\n")
            top_sources = sorted(
                self.stats['source_counts'].items(), 
                key=lambda x: x[1], 
                reverse=True
            )[:10]
            for source, count in top_sources:
                f.write(f"- {source}: {count:,}\n")
            
            f.write(f"\n發(fā)現(xiàn)錯(cuò)誤數(shù)量: {len(self.stats['errors'])}\n")
        
        print(f"已保存統(tǒng)計(jì)摘要: {stats_path}")
 
# 使用示例
if __name__ == "__main__":
    # 配置日志目錄路徑
    LOG_DIRECTORY = "/var/log/myapp"
    
    # 初始化分析器
    analyzer = LogAnalyzer(LOG_DIRECTORY)
    
    # 執(zhí)行分析
    print("開(kāi)始日志分析...")
    analyzer.analyze_directory()
    
    # 生成報(bào)告
    print("\n生成分析報(bào)告...")
    analyzer.generate_reports()
    
    print("\n分析完成！所有報(bào)告已保存至:", analyzer.output_dir)

class LogAnalyzer:
    DEFAULT_PATTERNS = {
        'timestamp': r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})',
        'level': r'(?P<level>DEBUG|INFO|WARNING|ERROR|CRITICAL)',
        'message': r'(?P<message>.*)',
        'source': r'(?P<source>\w+\.\w+)'
    }
    
    def __init__(self, log_dir: str, output_dir: str = "log_analysis"):
        self.log_dir = log_dir
        self.output_dir = output_dir
        os.makedirs(self.output_dir, exist_ok=True)
        
        self.patterns = {
            name: re.compile(pattern) 
            for name, pattern in self.DEFAULT_PATTERNS.items()
        }
        
        self.stats = {
            'total_lines': 0,
            'level_counts': {},
            'source_counts': {},
            'errors': [],
            'timeline': []
        }

def _parse_line(self, line: str) -> Optional[Dict[str, str]]:
    combined_pattern = re.compile(
        r'^{timestamp}\s+{level}\s+\[{source}\]\s+{message}$'.format(
            **self.DEFAULT_PATTERNS
        )
    )
    
    match = combined_pattern.match(line.strip())
    if match:
        return match.groupdict()
    return None

def _read_log_file(self, filepath: str) -> List[str]:
    if filepath.endswith('.gz'):
        with gzip.open(filepath, 'rt', encoding='utf-8') as f:
            return f.readlines()
    else:
        with open(filepath, 'r', encoding='utf-8') as f:
            return f.readlines()
 
def analyze_file(self, filepath: str):
    lines = self._read_log_file(filepath)
    filename = os.path.basename(filepath)
    
    for line in tqdm(lines, desc=f"分析 {filename}"):
        self.stats['total_lines'] += 1
        parsed = self._parse_line(line)
        
        if not parsed:
            continue
        # ...分析邏輯...

# 在analyze_file方法中
try:
    dt = parser.parse(parsed['timestamp'])
    self.stats['timeline'].append({
        'timestamp': dt,
        'level': parsed['level'],
        'source': parsed['source']
    })
except (ValueError, KeyError):
    pass
 
# 在generate_reports方法中
timeline_df = pd.DataFrame(self.stats['timeline'])
timeline_df.set_index('timestamp', inplace=True)

def _plot_level_distribution(self):
    levels = list(self.stats['level_counts'].keys())
    counts = list(self.stats['level_counts'].values())
    
    plt.figure(figsize=(10, 6))
    bars = plt.bar(levels, counts, color=['green', 'blue', 'orange', 'red', 'purple'])
    
    # 添加數(shù)值標(biāo)簽
    for bar in bars:
        height = bar.get_height()
        plt.text(bar.get_x() + bar.get_width()/2., height,
                f'{height:,}', ha='center', va='bottom')
    # ...保存圖片...

def add_log_format(self, name: str, pattern: str):
    """添加自定義日志格式"""
    try:
        self.patterns[name] = re.compile(pattern)
    except re.error as e:
        print(f"無(wú)效的正則表達(dá)式: {pattern} - {str(e)}")
 
def auto_detect_format(self, sample_lines: List[str]) -> bool:
    """自動(dòng)檢測(cè)日志格式"""
    common_formats = [
        (r'^(?P<timestamp>.+?) (?P<level>\w+) (?P<message>.+)$', "格式A"),
        (r'^\[(?P<timestamp>.+?)\] \[(?P<level>\w+)\] (?P<source>\w+) - (?P<message>.+)$', "格式B")
    ]
    
    for pattern, name in common_formats:
        matched = 0
        for line in sample_lines[:10]:  # 檢查前10行
            if re.match(pattern, line.strip()):
                matched += 1
        
        if matched >= 8:  # 80%匹配則認(rèn)為成功
            self.add_log_format(name, pattern)
            return True
    return False

def detect_anomalies(self, window_size: int = 60, threshold: int = 10):
    """檢測(cè)異常錯(cuò)誤爆發(fā)"""
    df = pd.DataFrame(self.stats['timeline'])
    error_df = df[df['level'].isin(['ERROR', 'CRITICAL'])]
    
    # 按分鐘統(tǒng)計(jì)錯(cuò)誤數(shù)
    error_counts = error_df.resample('1T', on='timestamp').size()
    
    # 使用滑動(dòng)窗口檢測(cè)異常
    rolling_mean = error_counts.rolling(window=window_size).mean()
    anomalies = error_counts[error_counts > (rolling_mean + threshold)]
    
    if not anomalies.empty:
        report = "\n".join(
            f"{ts}: {count} 個(gè)錯(cuò)誤 (平均: {rolling_mean[ts]:.1f})"
            for ts, count in anomalies.items()
        )
        print(f"檢測(cè)到異常錯(cuò)誤爆發(fā):\n{report}")
        
        # 保存異常報(bào)告
        with open(os.path.join(self.output_dir, 'anomalies.txt'), 'w') as f:
            f.write(report)

def handle_rotated_logs(self):
    """處理輪轉(zhuǎn)的日志文件"""
    for root, _, files in os.walk(self.log_dir):
        for file in files:
            if re.match(r'.*\.[0-9]+(\.gz)?$', file):  # 匹配輪轉(zhuǎn)文件如.log.1, .log.2.gz
                filepath = os.path.join(root, file)
                self.analyze_file(filepath)

from concurrent.futures import ProcessPoolExecutor
 
def parallel_analyze(self):
    log_files = self._find_log_files()
    with ProcessPoolExecutor() as executor:
        list(tqdm(executor.map(self.analyze_file, log_files), total=len(log_files)))

import unittest
import tempfile
import shutil
from pathlib import Path
 
class TestLogAnalyzer(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.test_dir = Path(tempfile.mkdtemp())
        cls.sample_log = cls.test_dir / "test.log"
        
        # 創(chuàng)建測(cè)試日志文件
        with open(cls.sample_log, 'w') as f:
            f.write("2023-01-01 12:00:00,123 INFO [app.core] System started\n")
            f.write("2023-01-01 12:00:01,456 ERROR [app.db] Connection failed\n")
    
    def test_parser(self):
        analyzer = LogAnalyzer(self.test_dir)
        parsed = analyzer._parse_line("2023-01-01 12:00:00,123 INFO [app.core] Test message")
        self.assertEqual(parsed['level'], 'INFO')
        self.assertEqual(parsed['source'], 'app.core')
    
    def test_analysis(self):
        analyzer = LogAnalyzer(self.test_dir)
        analyzer.analyze_file(self.sample_log)
        self.assertEqual(analyzer.stats['total_lines'], 2)
        self.assertEqual(analyzer.stats['level_counts']['INFO'], 1)
        self.assertEqual(analyzer.stats['level_counts']['ERROR'], 1)
    
    @classmethod
    def tearDownClass(cls):
        shutil.rmtree(cls.test_dir)
 
if __name__ == '__main__':
    unittest.main()